Release v0.1.3

opsi/lifespan/lifespan.py

@@ -9,22 +9,24 @@
 from opsi.manager import Program
 from opsi.manager.manager_schema import ModulePath
 from opsi.util.concurrency import AsyncThread, ShutdownThread
-from opsi.util.networking import choose_port, get_nt_server
+from opsi.util.networking import choose_port
 from opsi.util.path import join
 from opsi.util.persistence import Persistence
 from opsi.webserver import WebServer
 from opsi.webserver.serialize import import_nodetree
 
 from .webserverthread import WebserverThread
 
-# import optional dependencies
 try:
     from networktables import NetworkTables
 
     NT_AVAIL = True
 except ImportError:
     NT_AVAIL = False
 
+
+# import optional dependencies
+
 try:
     from pystemd.systemd1 import Unit
 
@@ -56,14 +58,6 @@ def register_modules(program, module_path):
         program.manager.register_module(ModulePath(moddir, path))
 
 
-def init_networktables(network):
-    if network.nt_client:
-        addr = get_nt_server(network)
-        NetworkTables.startClient(addr)
-    else:
-        NetworkTables.startServer()
-
-
 class Lifespan:
     PORTS = (80, 8000)
     NT_AVAIL = NT_AVAIL
@@ -83,6 +77,7 @@ def __init__(self, args, *, catch_signal=False, load_persist=True, timeout=10):
         self.persist = Persistence(path=args.persist) if load_persist else None
 
         if catch_signal:
+            self.signalCount = 0
             signal.signal(signal.SIGINT, self.catch_signal)
             signal.signal(signal.SIGTERM, self.catch_signal)
 
@@ -104,8 +99,6 @@ def using_systemd(self):
         return self._systemd
 
     def make_threads(self):
-        if self.NT_AVAIL and self.persist.network.nt_enabled:
-            init_networktables(self.persist.network)
         program = Program(self)
 
         path = opsi.__file__
@@ -146,6 +139,9 @@ def main_loop(self):
 
     def catch_signal(self, signum, frame):
         self.shutdown()
+        self.signalCount += 1
+        if self.signalCount >= 2:
+            self.terminate()
 
     def terminate(self):
         LOGGER.critical("OpenSight failed to close gracefully! Terminating...")
@@ -158,8 +154,6 @@ def shutdown_threads(self):
                     self.timer.start()
                 except RuntimeError:
                     pass
-        if NT_AVAIL and self.persist.network.nt_enabled:
-            NetworkTables.shutdown()
         LOGGER.info("Waiting for threads to shut down...")
         self.event.set()
 

opsi/manager/cvwrapper.py

@@ -1,3 +1,5 @@
+import math
+
 import cv2
 
 from .types import *
@@ -71,6 +73,60 @@ def hsv_threshold(img: Mat, hue: "Range", sat: "Range", lum: "Range") -> MatBW:
     return cv2.inRange(cv2.cvtColor(img.mat, cv2.COLOR_BGR2HSV), *ranges).view(MatBW)
 
 
+def v_threshold(img: Mat, val: "Range") -> MatBW:
+    """
+       Args:
+           img: Mat
+           val: Value range (min, max) (0 - 255)
+       Returns:
+           Black+White Mat
+       """
+    return cv2.inRange(img.mat, val[0], val[1]).view(MatBW)
+
+
+def hough_circles(
+    img: Mat,
+    dp: int,
+    min_dist: int,
+    param1: int,
+    param2: int,
+    min_radius: int,
+    max_radius: int,
+) -> "Circles":
+    return cv2.HoughCircles(
+        img,
+        method=cv2.HOUGH_GRADIENT,
+        dp=dp,
+        minDist=min_dist,
+        param1=param1,
+        param2=param2,
+        minRadius=min_radius,
+        maxRadius=max_radius,
+    )
+
+
+def hough_lines(
+    img: Mat,
+    rho: int,
+    threshold: int,
+    min_length: int,
+    max_gap: int,
+    theta: float = math.pi / 180.0,
+) -> "Segments":
+    return cv2.HoughLinesP(
+        img,
+        rho=rho,
+        theta=theta,
+        threshold=threshold,
+        minLineLength=min_length,
+        maxLineGap=max_gap,
+    )
+
+
+def canny(img: Mat, threshold_lower, threshold_upper) -> MatBW:
+    return cv2.Canny(img, threshold_lower, threshold_upper)
+
+
 ERODE_DILATE_CONSTS = {
     "kernel": None,
     "anchor": (-1, -1),
@@ -159,3 +215,11 @@ def joinBW(img1: MatBW, img2: MatBW) -> MatBW:
 
 def greyscale(img: Mat) -> Mat:
     return cv2.cvtColor(img.mat, cv2.COLOR_BGR2GRAY).view(Mat).mat
+
+
+def bgr_to_hsv(img: Mat) -> Mat:
+    return cv2.cvtColor(img.mat, cv2.COLOR_BGR2HSV)
+
+
+def abs_diff(img: Mat, scalar: ndarray):
+    return cv2.absdiff(img, scalar)

opsi/manager/manager.py

@@ -99,6 +99,8 @@ def register_module(self, path: ModulePath):
             hook = hooks_tuple[0][1]
             self.hooks[info.package] = hook
             setattr(hook, "pipeline", self.pipeline)
+            setattr(hook, "persist", self.pipeline.program.lifespan.persist)
+            hook.startup()
         elif len(hooks_tuple) > 1:
             LOGGER.error(f"Only one Hook per module allowed: {info.package}")
             return
@@ -118,3 +120,7 @@ def register_module(self, path: ModulePath):
             self.funcs[func.type] = func
 
         self.modules[info.package] = ModuleItem(info, funcs)
+
+    def shutdown(self):
+        for hook in self.hooks.values():
+            hook.shutdown()

opsi/manager/manager_schema.py

@@ -1,5 +1,4 @@
 import logging
-
 from dataclasses import Field, dataclass, fields, is_dataclass
 from typing import (
     Any,
@@ -170,13 +169,72 @@ def __init__(self, visible=True):
         self.visible = visible
         self.app = Router()
         self.url = ""  # will be replaced during webserver init
-
-    def cancel_dependents(self):
+        self.cache = {"skip": {}, "deps": {}}
+        self.listeners = {"startup": set(), "shutdown": set(), "pipeline_update": set()}
+        self.lastPipeline = None
+
+    def update_cache(self):
+        if not self.lastPipeline == self.pipeline.nodes:
+            self.cache = {"skip": {}, "deps": {}}
+        self.lastPipeline = self.pipeline.nodes
+
+    def get_skips(self, node):
+        self.update_cache()
+        skip = self.cache["skip"].get(node)
+        if skip is None:
+            skip = self.pipeline.get_dependents(node)
+            self.cache["skip"][node] = skip
+        return skip
+
+    def get_output_deps(self, node, output):
+        self.update_cache()
+
+        if node not in self.cache["deps"]:
+            self.cache["deps"][node] = {}
+
+        deps = self.cache["deps"][node].get(output)
+
+        if deps is None:
+            deps = []
+            for i in self.pipeline.nodes.values():
+                for link in i.inputLinks.values():
+                    if link.node is node and link.name == output:
+                        deps.append(i)
+            self.cache["deps"][node][output] = deps
+
+        return deps
+
+    def cancel_node(self, node):
         try:
-            self.pipeline.cancel_dependents(self.pipeline.current)
+            # reset path cache if pipeline has changed
+            skip = self.get_skips(node)
+            self.pipeline.cancel_nodes(skip)
         except:
             raise ValueError("Pipeline not available! Cannot cancel dependents.")
 
+    def cancel_current(self):
+        self.cancel_node(self.pipeline.current)
+
+    def cancel_output(self, output: str):
+        node = self.pipeline.current
+        deps = self.get_output_deps(node, output)
+        for dep in deps:
+            self.cancel_node(dep)
+
+    def add_listener(self, event: str, function: callable):
+        self.listeners[event].add(function)
+
+    def remove_listener(self, event: str, function: callable):
+        self.listeners[event].discard(function)
+
+    def startup(self):
+        for func in self.listeners["startup"]:
+            func()
+
+    def shutdown(self):
+        for func in self.listeners["shutdown"]:
+            func()
+
 
 def ishook(hook):
     return isinstance(hook, Hook)

opsi/manager/netdict.py

@@ -22,6 +22,7 @@
     List[str],
 ]
 
+_missing = object()
 
 class NetworkDict:
     def __init__(self, table: str, networktable: NetworkTablesInstance = NetworkTables):
@@ -65,11 +66,11 @@ def get_subtable(self, table: str) -> "NetworkDict":
 
     # Getters and setters
 
-    def get(self, name: str, default: NT_TYPES = None) -> NT_TYPES:
+    def get(self, name: str, default: NT_TYPES = _missing) -> NT_TYPES:
         entry = self._get_entry(name)
 
         if entry is None:
-            if default is not None:  # If a default is specified
+            if default is not _missing:  # If a default is specified
                 return default
 
             raise KeyError(name)

opsi/manager/pipeline.py

@@ -2,9 +2,9 @@
 import time
 from dataclasses import fields
 from itertools import chain
+from queue import deque
 from typing import Any, Dict, List, NamedTuple, Optional, Set, Type
 from uuid import UUID
-from queue import deque
 
 from toposort import toposort
 
@@ -142,14 +142,14 @@ def mainloop(self):
                 # avoid clogging logs with errors
                 time.sleep(0.5)
 
-    def cancel_dependents(self, node):
+    def get_dependents(self, node):
         visited = set()
         queue = deque()
         path = {}
 
         # First, add all side effect nodes to queue
-        for id, node in self.nodes.items():
-            if node.func_type.has_sideeffect:
+        for id, i in self.nodes.items():
+            if i.func_type.has_sideeffect:
                 queue.append(id)
 
         # Then, do a DFS over queue, adding all reachable nodes to visited
@@ -159,26 +159,31 @@ def cancel_dependents(self, node):
 
             if id in visited:
                 continue
-
             if id == node.id:
                 break
 
             for input in self.nodes[id].inputLinks.values():
                 link = input
-
                 if link is None:
                     continue
-
                 queue.append(link.node.id)
                 path[link.node] = self.nodes[id]
 
-        # Iterate through path and skip all nodes which were visited
         pathTemp = node
+        skip_nodes = []
         while pathTemp is not None:
             # Don't skip supplied node, since that would be applied next run
-            if pathTemp is not node:
-                pathTemp.skip = True
+            # if pathTemp is not node:
+            skip_nodes.append(pathTemp)
             pathTemp = path.get(pathTemp)
+        return skip_nodes
+
+    def cancel_nodes(self, nodes):
+        for node in nodes:
+            node.skip = True
+
+    # def cancel_dependents(self, node, path):
+    # Iterate through path and skip all nodes which were visited
 
     def create_node(self, func: Type[Function], uuid: UUID):
         """
@@ -214,10 +219,12 @@ def prune_nodetree(self, new_node_ids):
         new_node_ids = set(new_node_ids)
         removed = old_node_ids - new_node_ids
 
+        self.clear()
         # remove deleted nodes
         for uuid in removed:
             try:
                 self.nodes[uuid].dispose()
+                del self.adjList[self.nodes[uuid]]
                 del self.nodes[uuid]
             except KeyError:
                 pass

opsi/manager/program.py

@@ -46,4 +46,5 @@ def mainloop(self, shutdown):
             task.run()  # won't send exceptions because runs in seperate thead
         LOGGER.info("Program main loop is shutting down...")
         self.pipeline.clear()
+        self.manager.shutdown()
         self.shutdown.clear()

opsi/manager/types.py

@@ -61,6 +61,18 @@ def create(self, min, max):
 # Making new classes allows me to do simple equality testing
 
 
+class Circles(ndarray):
+    pass
+
+
+class Segments(ndarray):
+    pass
+
+
+class Lines(ndarray):
+    pass
+
+
 class Contour(ndarray):
     pass