""" Chemical Compliance Review Flask Application This application serves as an External Action for Signals Notebook, allowing users to review and validate chemical reactions for compliance. Usage: python app.py Then navigate to: http://localhost:5000/review?__eid=chemicalDrawing:YOUR_CHEMICAL_DRAWING_EID Templates: This application uses a single Jinja2 template (templates/index.html) that adjusts its content based on the current state (review, already_compliant, result). """ from flask import Flask, request, render_template import requests app = Flask(__name__) # Configuration - Replace with your values CONFIG = { "base_url": "https://YOUR_TENANT.signalsnotebook.com/api/rest/v1.0", "api_key": "YOUR_API_KEY" } HEADERS = { "x-api-key": CONFIG["api_key"], "Content-Type": "application/vnd.api+json", "Accept": "application/vnd.api+json" } # Compliance status options with Unicode indicators COMPLIANCE_OPTIONS = { "compliant": {"symbol": "🟢", "label": "Compliant", "text": "🟢 Compliant"}, "warning": {"symbol": "🟡", "label": "Warning", "text": "🟡 Compliant with Warning"}, "non_compliant": {"symbol": "🔴", "label": "Non-Compliant", "text": "🔴 Non-Compliant"} } def get_stoichiometry(eid): """Fetch stoichiometry data for a chemical drawing.""" url = f"{CONFIG['base_url']}/stoichiometry/{eid}" response = requests.get(url, headers=HEADERS) response.raise_for_status() return response.json() def get_compliance(eid): """Fetch compliance data for a chemical drawing.""" url = f"{CONFIG['base_url']}/entities/{eid}/compliance" response = requests.get(url, headers=HEADERS) response.raise_for_status() return response.json() def update_stoichiometry_row(eid, row_id, compliance_key, status_text): """Update the compliance column for a stoichiometry row.""" url = f"{CONFIG['base_url']}/stoichiometry/{eid}/{row_id}" payload = { "data": { "attributes": { "values": { compliance_key: status_text } } } } response = requests.patch(url, headers=HEADERS, params={"force": "true"}, json=payload) response.raise_for_status() return response.json() def dismiss_external_warning(eid): """Dismiss the external checking warning.""" compliance = get_compliance(eid) self_hash = compliance["data"]["attributes"]["self"].get("hash") if not self_hash: return False url = f"{CONFIG['base_url']}/entities/{eid}/compliance" payload = { "data": { "attributes": { "external": { "hash": self_hash } } } } response = requests.patch(url, headers=HEADERS, params={"force": "true"}, json=payload) response.raise_for_status() return True def find_compliance_column_key(column_definitions, grid_type): """Find the key for the Compliance column in column definitions.""" columns = column_definitions.get(grid_type, []) for col in columns: if col.get("title", "").lower() == "compliance": return col.get("key") return None def is_already_compliant(eid): """Check if the chemical drawing is already compliant (hashes match).""" try: compliance = get_compliance(eid) attrs = compliance["data"]["attributes"] self_hash = attrs.get("self", {}).get("hash") external_hash = attrs.get("external", {}).get("hash") return self_hash and external_hash and self_hash == external_hash except: return False @app.route("/review") def review(): """Display the compliance review interface.""" eid = request.args.get("__eid") if not eid: return "Error: No entity ID provided. Expected ?__eid=chemicalDrawing:...", 400 # Check if force re-review is requested force_review = request.args.get("force", "").lower() == "true" try: # Check if already compliant (unless force review is requested) if not force_review and is_already_compliant(eid): return render_template("index.html", eid=eid, state="already_compliant") stoich_data = get_stoichiometry(eid) attributes = stoich_data["data"]["attributes"] reactants = attributes.get("reactants", []) products = attributes.get("products", []) return render_template( "index.html", eid=eid, state="review", reactants=reactants, products=products, options=COMPLIANCE_OPTIONS ) except Exception as e: return f"Error fetching data: {str(e)}", 500 @app.route("/submit", methods=["POST"]) def submit_review(): """Process the compliance review submission.""" eid = request.form.get("eid") if not eid: return "Error: No entity ID provided", 400 try: # Get stoichiometry data to find column keys and row IDs stoich_data = get_stoichiometry(eid) attributes = stoich_data["data"]["attributes"] # Get column definitions to find compliance column keys included = stoich_data.get("included", []) column_defs = {} for item in included: if item.get("type") == "columnDefinitions": column_defs = item.get("attributes", {}) break reactant_compliance_key = find_compliance_column_key(column_defs, "reactants") product_compliance_key = find_compliance_column_key(column_defs, "products") # Track compliance statuses all_statuses = [] reactants_updated = 0 products_updated = 0 # Process reactants reactants = attributes.get("reactants", []) for reactant in reactants: row_id = reactant.get("row_id") if row_id and reactant_compliance_key: status_key = request.form.get(f"reactant_{row_id}", "compliant") status_text = COMPLIANCE_OPTIONS[status_key]["text"] all_statuses.append(status_key) update_stoichiometry_row(eid, row_id, reactant_compliance_key, status_text) reactants_updated += 1 # Process products products = attributes.get("products", []) for product in products: row_id = product.get("row_id") if row_id and product_compliance_key: status_key = request.form.get(f"product_{row_id}", "compliant") status_text = COMPLIANCE_OPTIONS[status_key]["text"] all_statuses.append(status_key) update_stoichiometry_row(eid, row_id, product_compliance_key, status_text) products_updated += 1 # Determine overall compliance has_non_compliant = "non_compliant" in all_statuses has_warnings = "warning" in all_statuses all_compliant = all(s == "compliant" for s in all_statuses) if all_statuses else True # Dismiss warning if no non-compliant items warning_dismissed = False if not has_non_compliant: warning_dismissed = dismiss_external_warning(eid) return render_template( "index.html", eid=eid, state="result", all_compliant=all_compliant, has_warnings=has_warnings and not has_non_compliant, reactants_updated=reactants_updated, products_updated=products_updated, warning_dismissed=warning_dismissed ) except Exception as e: return f"Error processing review: {str(e)}", 500 if __name__ == "__main__": print("=" * 60) print("Chemical Compliance Review Server") print("=" * 60) print("") print("Test URL:") print(" http://localhost:5000/review?__eid=chemicalDrawing:YOUR_CHEMICAL_DRAWING_EID") print("") print("Template: ./templates/index.html") print("") print("=" * 60) app.run(debug=True, port=5000)