tsnkit.core package

Submodules

tsnkit.core._calculate module

Author: Chuanyu (skewcy@gmail.com) _calculate.py (c) 2023 Desc: description Created: 2023-10-08T04:36:18.116Z

tsnkit.core._common module

Author: Chuanyu (skewcy@gmail.com) _common.py (c) 2023 Desc: description Created: 2023-10-08T17:51:27.418Z

tsnkit.core._common._interface(name: str) → Any

tsnkit.core._common.benchmark(stream_path, network_path)

tsnkit.core._common.flatten(list2d: Sequence[Sequence[Any]]) → Sequence[Any]

tsnkit.core._common.parse_command_line_args()

tsnkit.core._common.parse_command_line_constants(parser: ArgumentParser)

tsnkit.core._config module

Author: Chuanyu (skewcy@gmail.com) _schedule.py (c) 2023 Desc: description Created: 2023-10-08T06:13:56.911Z

class tsnkit.core._config.Config

Bases: object

to_csv(name, path)

class tsnkit.core._config.Delay(init_list: List[List])

Bases: list

A list of [stream_id, frame_id, delay]

static format_delay_type(init_list: List[List]) → None

[Note]: This will modify the input list

Args:

init_list (List[List]): _description_

static is_valid_delay_format(init_list: List[List]) → bool

to_csv(path: str) → None

class tsnkit.core._config.GCL(init_list: List[List])

Bases: list

A list of [link, queue, start, end, cycle]

Args:

init_list (List[List[link, queue, start, end, cycle]]): [description]

static format_gcl_type(init_list: List[List]) → bool

static is_overlap(init_list: List[List]) → bool

static is_small_entry(init_list: List[List]) → bool

static is_valid_gcl_format(init_list: List[List]) → bool

to_csv(path: str) → None

class tsnkit.core._config.Queue(init_list: List[List])

Bases: list

A list of [stream_id, frame_id, link, queue]

Args:

init_list (List[List[stream_id, frame_id, link, queue]]): [description]

static format_queue_type(init_list: List[List]) → None

[Note]: This will modify the input list

Args:

init_list (List[List]): _description_

static is_valid_queue_format(init_list: List[List]) → bool

static is_valid_queue_range(init_list: List[List]) → bool

to_csv(path: str) → None

class tsnkit.core._config.Release(init_list: List[List])

Bases: list

A list of [stream_id, frame_id, release_time]

static format_release_type(init_list: List[List]) → None

[Note]: This will modify the input list

Args:

init_list (List[List]): _description_

static is_valid_release_format(init_list: List[List]) → bool

to_csv(path: str) → None

class tsnkit.core._config.Route(init_list: List[List])

Bases: list

A list of [stream_id, link] Each stream can contains several links as its route

format_route_type(init_list)

static is_valid_route_format(init_list: List[List]) → bool

static is_valid_route_logic(init_list: List[List]) → bool

to_csv(path)

class tsnkit.core._config.Size(iterable=(), /)

Bases: list

This i only used for FRAG based model now [NOTE]: No type check for

to_csv(path: str) → None

tsnkit.core._config.result_slot_to_ns(result: List[List[float]], col: List[int]) → bool

_summary_ [NOTE] it will modify the input result

tsnkit.core._constants module

Author: Chuanyu (skewcy@gmail.com) _constants.py (c) 2023 Desc: description Created: 2023-10-21T02:24:36.212Z

tsnkit.core._io module

Author: Chuanyu (skewcy@gmail.com) _io.py (c) 2023 Desc: description Created: 2023-10-08T06:13:41.041Z

class tsnkit.core._io.Result(value)

Bases: Enum

error = -1

schedulable = 1

unknown = 2

unschedulable = 0

class tsnkit.core._io.Statistics(name: str = '-', result: Result = Result.unknown, algo_time: float = 0, algo_mem: float = 0, extra_time: float = 0, extra_mem: float = 0)

Bases: object

content(name: str = '-') → None

Print the content of the output

header() → None

Print the header of the output

output_format = '| {:<13} | {:<13} | {:<6} | {:<10} | {:<10} | {:<10}'

to_list() → List[str | float]

Convert the statistics to a list

Returns:

List[Union[str, int]]: [name, result, algo_time, total_time, algo_mem, total_mem]

update(result: Result, algo_time: int, algo_mem: int, extra_time: int = 0, extra_mem: int = 0) → None

tsnkit.core._io.check_time_limit(func)

tsnkit.core._io.get_caller_name() → str

Used to be myname function

tsnkit.core._network module

Author: Chuanyu (skewcy@gmail.com) _network.py (c) 2023 Desc: description Created: 2023-10-08T06:13:46.561Z

class tsnkit.core._network.Link(*args, **kwargs)

Bases: int

Use simple link structure to improve the efficiency

The overload __hash__ and __eq__ function is used to achieve flexible interface. -> Use (0, 1), Link(src=0, dst=1), (Node(0), Node(1)) are equivalent in framework.

property dst

property name

property q_num

property rate

property src

property t_proc

property t_prop

property t_sync

class tsnkit.core._network.Network

Bases: object

add_link()

del_link()

property e_nodes: List[Node]

Return all end station nodes

Returns:

List[Node]: _description_

get_all_path(src: int | Node, dst: int | Node) → List[Path]

get_income_links(node: int | Node) → List[Link]

get_income_nodes(node: int | Node) → List[Node]

get_link(lid: int | Link | Tuple[int | Node, int | Node]) → Link

get_link_pairs(permute: bool = False) → List[Tuple[Link, Link]]

get_links() → List[Link]

get_node(nid: int | Node) → Node

get_nodes() → List[Node]

get_nodes_es() → List[Node]

Return all end station nodes

Returns:

List[Node]: _description_

get_nodes_sw() → List[Node]

Return all switch nodes

Returns:

List[Node]: _description_

get_outcome_links(node: int | Node) → List[Link]

get_outcome_nodes(node: int | Node) → List[Node]

get_shortest_path(src: int | Node, dst: int | Node) → Path

property links: List[Link]

property max_t_proc: int

property net_np

property net_nx

property nodes: List[Node]

property num_l: int

property num_n: int

property s_nodes: List[Node]

Return all switch nodes

Returns:

List[Node]: _description_

update_link()

class tsnkit.core._network.Node(*args, **kwargs)

Bases: int

property num_port

property sync_error

property type

class tsnkit.core._network.NodeType(value)

Bases: Enum

Sample enum class for node type

es = 1

sw = 0

class tsnkit.core._network.Path(path: List[int | Node | Tuple[int | Node, int | Node] | Link], network: Network)

Bases: object

A path can be either a node path or a link path Initialize with node path or link path, the other one will be automatically

No iterator interface to avoid misuse

get_in_link(node: int | Node) → Link | None

get_len_link() → int

get_len_node() → int

get_link(link: int | Link | Tuple[int | Node, int | Node]) → Link

get_next_link(link: int | Link | Tuple[int | Node, int | Node]) → Link | None

get_next_node(node: int | Node) → Node | None

get_out_link(node: int | Node) → Link | None

get_prev_link(link: int | Link | Tuple[int | Node, int | Node]) → Link | None

get_prev_node(node: int | Node) → Node | None

iter_link() → Iterator[Link]

iter_node() → Iterator[Node]

static link_path_to_node_path(link_path: List[Link], network: Network) → List[Node]

property links

property llen

property nlen

static node_path_to_link_path(node_path: List[Node], network: Network) → List[Link]

property nodes

static sort_links(links: List[Link | Tuple[int | Node, int | Node]]) → List[Link | Tuple[int | Node, int | Node]]

tsnkit.core._network.check_network_format(network: DataFrame)

tsnkit.core._network.load_network(path: str) → Network

tsnkit.core._stream module

Author: Chuanyu (skewcy@gmail.com) _stream.py (c) 2023 Desc: description Created: 2023-10-08T06:14:04.079Z

class tsnkit.core._stream.Stream(*args, **kwargs)

Bases: int

property deadline

property dst

property dst_mul

property first_link: Link

get_frame_indexes(lcm: int) → List[int]

get_link(link: int | Link | Tuple[int | Node, int | Node]) → Link

get_next_link(link: int | Link | Tuple[int | Node, int | Node]) → Link | None

get_next_node(node: int | Node) → Node | None

get_num_frames(lcm: int) → int

get_prev_link(link: int | Link | Tuple[int | Node, int | Node]) → Link | None

get_prev_node(node: int | Node) → Node | None

get_t_trans(link: int | Link | Tuple[int | Node, int | Node]) → int

is_in_path(link: int | Link | Tuple[int | Node, int | Node]) → bool

property jitter

property last_link: Link

property links: List[Link]

property period

property routing_path: Path

property size

property src

property t_trans: int

property t_trans_1g: int

Return max(transmission time on all links) Only used for uniform link rate assuming 1Gbps link rate. [NOTE]: This method it depreciated.

Returns:

int: _description_

class tsnkit.core._stream.StreamSet

Bases: object

get_frame_index_pairs(s1: int | Stream, s2: int | Stream)

Get all frame index pairs of two streams. LCM is calculated from s1.period and s2.period

Args:

s1 (Union[int, Stream]): _description_ s2 (Union[int, Stream]): _description_

Returns:

_type_: _description_

get_merged_links(s1: int | Stream, s2: int | Stream) → List[Tuple[Link, Link, Link]]

This function is often used in queue isolation constraint. Example:

Path1 = (a, b), (b, c) Path2 = (d, b), (b, c)

return [((a, b), (d, b), (b, c))]

Returns:

List[Tuple[Link1, Link2, Link3]]: Link1 and Link2 are the previous links of Link3. Link1, Link3 are in s1.Path Link2, Link3 are in s2.Path.

get_next_link(stream: int | Stream, link: int | Link | Tuple[int | Node, int | Node]) → Link | None

get_next_node(stream: int | Stream, node: int | Node) → Node | None

get_pairs(permute: bool = False) → List[Tuple[Stream, Stream]]

get_pairs_on_link(link: int | Link | Tuple[int | Node, int | Node], permute: bool = False) → List[Tuple[Stream, Stream]]

Return all pairs of streams that share the same link

Args:

link (FlexLink): _description_ permute (bool, optional): _description_. Defaults to False. -> only return (i, j) where i < j

Raises:

Exception: _description_

Returns:

List[Tuple[Stream, Stream]]: _description_

get_prev_link(stream: int | Stream, link: int | Link | Tuple[int | Node, int | Node]) → Link | None

get_prev_node(stream: int | Stream, node: int | Node) → Node | None

get_shared_links(stream1: int | Stream, stream2: int | Stream) → List[Link]

get_stream(stream: int | Stream) → Stream

get_streams() → List[Stream]

get_streams_on_link(link: int | Link | Tuple[int | Node, int | Node]) → List[Stream]

get_t_trans(stream: int | Stream, link: int | Link | Tuple[int | Node, int | Node]) → int

is_route_valid(stream: int | Stream)

property lcm

property length: int

property num_frames: int

set_routing(stream: int | Stream, routing_path: Path) → None

_summary_

Args:

stream (_type_): Stream_id or Stream object routing_path (_type_): Must be Path Object

set_routings(routings: Dict[int | Stream, Path] | List[Path])

_summary_

Args:

routings (_type_): Must be either dict of {stream_id: Path} or list of [Path] where

property streams

tsnkit.core._stream.check_stream_format(stream_df: DataFrame)

tsnkit.core._stream.load_stream(path: str) → StreamSet

tsnkit.core._system module

Author: Chuanyu (skewcy@gmail.com) _system.py (c) 2023 Desc: description Created: 2023-10-08T06:14:11.998Z

tsnkit.core._system.find_files_with_prefix(directory: str, prefix: str)

tsnkit.core._system.init_output_folder(path: str) → None

tsnkit.core._system.is_timeout(thres: float) → bool

tsnkit.core._system.kill_if(main_proc: int, mem_limit: int, time_limit: int, sig: SynchronizedBase)

Kill the process if it uses more than mem memory or more than time seconds Args:

main_proc: the main process id mem_limit: the memory limit, uint: GB time_limit: the time limit, uint: seconds

async tsnkit.core._system.kill_process(proc: Process, time_limit: float)

tsnkit.core._system.mem_log() → float

tsnkit.core._system.oom_manager(name: str) → Popen

tsnkit.core._system.time_log() → float

Module contents

Author: Chuanyu (skewcy@gmail.com) __init__.py (c) 2023 Desc: description Created: 2023-10-06T17:54:47.806Z