zodiac.streams.model_stream 

 1#  # # <!-- // /*  SPDX-License-Identifier: MPL-2.0*/ -->
 2#  # # <!-- // /*  d a r k s h a p e s */ -->
 3
 4from typing import List, Tuple
 5
 6from toga.sources import Source
 7
 8nfo = print
 9
10
11class ModelStream(Source):
12    async def model_graph(self) -> None:
13        """Build an intent graph from models using the IntentProcessor class"""
14        from zodiac.graph import IntentProcessor
15
16        self._graph = {}
17        self._graph = IntentProcessor()
18        await self._graph.calc_graph()
19
20    async def show_edges(self, target: bool = False) -> List[str]:
21        """Retrieve and sort edges from the intent graph.\n
22        :param target: If True, sorts based on the second element of each edge pair; defaults to False.
23        :return: A sorted list of unique elements from the edge pairs."""
24
25        if self._graph.intent_graph:
26            edge_pairs = list(self._graph.intent_graph.edges)
27            if edge_pairs:
28                pair = 0 if not target else 1
29                seen = []
30                for edge in edge_pairs:
31                    if edge[pair] not in seen:
32                        seen.append(edge[pair])
33                seen.sort(key=len)
34                return seen
35
36    async def trace_models(self, mode_in: str, mode_out: str) -> List[Tuple[str, int]]:
37        """Trace model path through input to output mode, then updates the internal model list..\n
38        :param mode_in: The input mode for tracing.
39        :param mode_out: The output mode for tracing.
40        :return: A list of traced models."""
41
42        from nnll.monitor.file import dbuq
43
44        self._graph.set_path(mode_in=mode_in, mode_out=mode_out)
45        self._graph.set_registry_entries()
46        nfo(f"calculated : {self._graph.coord_path}")
47        dbuq(f"calculated : {self._graph.coord_path} {self._graph.registry_entries}")
48        self._models = self._graph.models
49        return self._models
50
51    async def chart_path(self) -> List[str]:
52        """Return hop names of current path\n
53        :return: List of [x,y,z] node names along the chosen path
54        """
55        return self._graph.coord_path
56
57    def __len__(self):
58        return len(list(self._models()))
59
60    def __getitem__(self, index):
61        return self._models()[index]
62
63    def index(self, entry):
64        return self._models().index(entry)
65
66    async def clear(self):
67        self._models = []
68        self.notify("clear")
def nfo(*args, sep=' ', end='\n', file=None, flush=False):

Prints the values to a stream, or to sys.stdout by default.

sep string inserted between values, default a space. end string appended after the last value, default a newline. file a file-like object (stream); defaults to the current sys.stdout. flush whether to forcibly flush the stream.

class ModelStream(toga.sources.base.Source): 
12class ModelStream(Source):
13    async def model_graph(self) -> None:
14        """Build an intent graph from models using the IntentProcessor class"""
15        from zodiac.graph import IntentProcessor
16
17        self._graph = {}
18        self._graph = IntentProcessor()
19        await self._graph.calc_graph()
20
21    async def show_edges(self, target: bool = False) -> List[str]:
22        """Retrieve and sort edges from the intent graph.\n
23        :param target: If True, sorts based on the second element of each edge pair; defaults to False.
24        :return: A sorted list of unique elements from the edge pairs."""
25
26        if self._graph.intent_graph:
27            edge_pairs = list(self._graph.intent_graph.edges)
28            if edge_pairs:
29                pair = 0 if not target else 1
30                seen = []
31                for edge in edge_pairs:
32                    if edge[pair] not in seen:
33                        seen.append(edge[pair])
34                seen.sort(key=len)
35                return seen
36
37    async def trace_models(self, mode_in: str, mode_out: str) -> List[Tuple[str, int]]:
38        """Trace model path through input to output mode, then updates the internal model list..\n
39        :param mode_in: The input mode for tracing.
40        :param mode_out: The output mode for tracing.
41        :return: A list of traced models."""
42
43        from nnll.monitor.file import dbuq
44
45        self._graph.set_path(mode_in=mode_in, mode_out=mode_out)
46        self._graph.set_registry_entries()
47        nfo(f"calculated : {self._graph.coord_path}")
48        dbuq(f"calculated : {self._graph.coord_path} {self._graph.registry_entries}")
49        self._models = self._graph.models
50        return self._models
51
52    async def chart_path(self) -> List[str]:
53        """Return hop names of current path\n
54        :return: List of [x,y,z] node names along the chosen path
55        """
56        return self._graph.coord_path
57
58    def __len__(self):
59        return len(list(self._models()))
60
61    def __getitem__(self, index):
62        return self._models()[index]
63
64    def index(self, entry):
65        return self._models().index(entry)
66
67    async def clear(self):
68        self._models = []
69        self.notify("clear")

A base class for data sources, providing an implementation of data notifications.

async def model_graph(self) -> None: 
13    async def model_graph(self) -> None:
14        """Build an intent graph from models using the IntentProcessor class"""
15        from zodiac.graph import IntentProcessor
16
17        self._graph = {}
18        self._graph = IntentProcessor()
19        await self._graph.calc_graph()

Build an intent graph from models using the IntentProcessor class

async def show_edges(self, target: bool = False) -> List[str]: 
21    async def show_edges(self, target: bool = False) -> List[str]:
22        """Retrieve and sort edges from the intent graph.\n
23        :param target: If True, sorts based on the second element of each edge pair; defaults to False.
24        :return: A sorted list of unique elements from the edge pairs."""
25
26        if self._graph.intent_graph:
27            edge_pairs = list(self._graph.intent_graph.edges)
28            if edge_pairs:
29                pair = 0 if not target else 1
30                seen = []
31                for edge in edge_pairs:
32                    if edge[pair] not in seen:
33                        seen.append(edge[pair])
34                seen.sort(key=len)
35                return seen

Retrieve and sort edges from the intent graph.

Parameters
  • target: If True, sorts based on the second element of each edge pair; defaults to False.
Returns

A sorted list of unique elements from the edge pairs.

async def trace_models(self, mode_in: str, mode_out: str) -> List[Tuple[str, int]]: 
37    async def trace_models(self, mode_in: str, mode_out: str) -> List[Tuple[str, int]]:
38        """Trace model path through input to output mode, then updates the internal model list..\n
39        :param mode_in: The input mode for tracing.
40        :param mode_out: The output mode for tracing.
41        :return: A list of traced models."""
42
43        from nnll.monitor.file import dbuq
44
45        self._graph.set_path(mode_in=mode_in, mode_out=mode_out)
46        self._graph.set_registry_entries()
47        nfo(f"calculated : {self._graph.coord_path}")
48        dbuq(f"calculated : {self._graph.coord_path} {self._graph.registry_entries}")
49        self._models = self._graph.models
50        return self._models

Trace model path through input to output mode, then updates the internal model list..

Parameters
  • mode_in: The input mode for tracing.
  • mode_out: The output mode for tracing.
Returns

A list of traced models.

async def chart_path(self) -> List[str]: 
52    async def chart_path(self) -> List[str]:
53        """Return hop names of current path\n
54        :return: List of [x,y,z] node names along the chosen path
55        """
56        return self._graph.coord_path

Return hop names of current path

Returns

List of [x,y,z] node names along the chosen path

def index(self, entry): 
64    def index(self, entry):
65        return self._models().index(entry)
async def clear(self): 
67    async def clear(self):
68        self._models = []
69        self.notify("clear")