Update genesis/safety.py

genesis/safety.py

@@ -1,36 +1,58 @@
-from __future__ import annotations
-import re
-from pydantic import BaseModel
-from typing import List
+# genesis/visualization.py
+"""
+Visualization tools for GENESIS-AI
+Uses Neo4j to query and build graphs.
+"""
 
-DUAL_USE_PATTERNS = [
-    r"step-?by-?step",
-    r"protocol",
-    r"wet[- ]?lab",
-    r"culture\s*conditions",
-    r"viral\s*vector",
-    r"pathogen",
-    r"gain[- ]of[- ]function",
-    r"increase\s*virulence",
-    r"synthesis\s*of\s*toxin",
-    r"biosafety\s*level\s*(2|3|4)",
-    r"kill\s*curve",
-    r"CFU|colony\s*forming\s*units",
-]
+import os
+import networkx as nx
+import matplotlib.pyplot as plt
+from py2neo import Graph
 
-class SafetyDecision(BaseModel):
-    allowed: bool
-    rationale: str
-    redactions: List[str] = []
+NEO4J_URI = os.getenv("NEO4J_URI")
+NEO4J_USER = os.getenv("NEO4J_USER")
+NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD")
 
-class SafetyGuard:
-    """Blocks operational/dual‑use outputs; allows high‑level literature review only."""
-    def gate(self, text: str) -> SafetyDecision:
-        hits = [p for p in DUAL_USE_PATTERNS if re.search(p, text, flags=re.I)]
-        if hits:
-            return SafetyDecision(
-                allowed=False,
-                rationale="Operational/dual‑use intent detected. Only high‑level review permitted.",
-                redactions=hits,
-            )
-        return SafetyDecision(allowed=True, rationale="High‑level research intent.")
+graph_db = Graph(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
+
+def generate_pathway_graph(pathway_name):
+    """Generate a metabolic pathway graph from Neo4j data."""
+    query = f"""
+    MATCH (p:Pathway {{name: '{pathway_name}'}})-[r:INVOLVES]->(m:Molecule)
+    RETURN p, r, m
+    """
+    data = graph_db.run(query).data()
+    G = nx.Graph()
+    for record in data:
+        p_name = record["p"]["name"]
+        m_name = record["m"]["name"]
+        G.add_node(p_name, color="green")
+        G.add_node(m_name, color="orange")
+        G.add_edge(p_name, m_name)
+    return draw_graph(G)
+
+def generate_funding_network(company_name):
+    """Generate company-investor funding network."""
+    query = f"""
+    MATCH (c:Company {{name: '{company_name}'}})-[r:FUNDED_BY]->(i:Investor)
+    RETURN c, r, i
+    """
+    data = graph_db.run(query).data()
+    G = nx.Graph()
+    for record in data:
+        c_name = record["c"]["name"]
+        i_name = record["i"]["name"]
+        G.add_node(c_name, color="green")
+        G.add_node(i_name, color="blue")
+        G.add_edge(c_name, i_name)
+    return draw_graph(G)
+
+def draw_graph(G):
+    """Draw networkx graph as PNG."""
+    pos = nx.spring_layout(G)
+    colors = [d.get("color", "grey") for _, d in G.nodes(data=True)]
+    plt.figure(figsize=(8, 6))
+    nx.draw(G, pos, with_labels=True, node_color=colors, node_size=1500, font_size=10)
+    plt.savefig("graph.png")
+    plt.close()
+    return "graph.png"