Mysql

cnceye/edge/find.py

@@ -22,6 +22,29 @@ def find_edge(filepath: str, minimal_diff: float = 5.0):
         previous_distance = distance
 
 
+def find_edges(minimal_diff: float = 5.0, mysql_config=MYSQL_CONFIG):
+    cnx = mysql.connector.connect(**mysql_config, database="coord")
+    cursor = cnx.cursor()
+    query = "SELECT * FROM sensor"
+    cursor.execute(query)
+    rows = cursor.fetchall()
+    previous_distance = ""
+    edges = []
+    for row in rows:
+        # x, y, z, distance
+        distance = row[4]
+        if check_if_edge_is_found(distance, previous_distance, minimal_diff):
+            edges.append(row)
+        previous_distance = distance
+
+    # remove the starting point
+    edges.pop(0)
+
+    cursor.close()
+    cnx.close()
+    return edges
+
+
 def find_edges_from_sqlite(database_path: str, minimal_diff: float = 5.0):
     conn = sqlite3.connect(database_path)
     cur = conn.cursor()
@@ -80,8 +103,8 @@ def find_lines(filepath: str, edge_count: int, minimal_diff: float = 5.0):
     return lines
 
 
-def get_edge_data():
-    cnx = mysql.connector.connect(**MYSQL_CONFIG, database="coord")
+def get_edge_data(mysql_config=MYSQL_CONFIG):
+    cnx = mysql.connector.connect(**mysql_config, database="coord")
     cursor = cnx.cursor()
     query = "SELECT id,side_id,x,y,z FROM edge"
     cursor.execute(query)
@@ -110,8 +133,8 @@ def identify_close_edge(edges, measured_edges, distance_threshold=2.5):
     return update_list
 
 
-def add_measured_edge_coord(edge_list: list):
-    cnx = mysql.connector.connect(**MYSQL_CONFIG, database="coord")
+def add_measured_edge_coord(edge_list: list, mysql_config=MYSQL_CONFIG):
+    cnx = mysql.connector.connect(**mysql_config, database="coord")
     cursor = cnx.cursor()
     insert_query = "UPDATE edge SET rx = %s, ry = %s, rz = %s WHERE id = %s"
     try:
@@ -121,3 +144,13 @@ def add_measured_edge_coord(edge_list: list):
     cnx.commit()
     cursor.close()
     cnx.close()
+
+
+def process_edges():
+    """
+    Identify the edges from the sensor data and add the coordinates to the database
+    """
+    measured_edges = find_edges()
+    edge_data = get_edge_data()
+    update_list = identify_close_edge(edge_data, measured_edges)
+    add_measured_edge_coord(update_list)

mysql/init.sql

@@ -62,4 +62,24 @@ CREATE TABLE IF NOT EXISTS `side` (
   `z1` FLOAT NOT NULL,
   `pair_id` int(11) unsigned,
   PRIMARY KEY (`id`)
+) ENGINE=InnoDB  DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci AUTO_INCREMENT=1;
+
+CREATE TABLE IF NOT EXISTS `sensor` (
+  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
+  `x` FLOAT NOT NULL,
+  `y` FLOAT NOT NULL,
+  `z` FLOAT NOT NULL,
+  `distance` FLOAT NOT NULL,
+  `timestamp` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB  DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci AUTO_INCREMENT=1;
+
+
+CREATE TABLE IF NOT EXISTS `process` (
+  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
+  `status` varchar(255) NOT NULL,
+  `error` varchar(255),
+  `start` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  `end` TIMESTAMP,
+  PRIMARY KEY (`id`)
 ) ENGINE=InnoDB  DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci AUTO_INCREMENT=1;

mysql/reset.sql

@@ -1,3 +1,6 @@
 TRUNCATE `point`;
 TRUNCATE `line`;
-TRUNCATE `arc`;
+TRUNCATE `arc`;
+TRUNCATE `edge`;
+TRUNCATE `side`;
+TRUNCATE `sensor`;

pyproject.toml

@@ -1,9 +1,12 @@
 [tool.poetry]
 name = "cnceye"
-version = "0.3.0"
-description = ""
+version = "0.3.1"
+description = "CMM python library"
+license = "MIT"
 authors = ["yuichiroaoki <[email protected]>"]
 readme = "README.md"
+repository = "https://github.com/OpenCMM/cnceye"
+keywords = ["cnc", "cmm", "milling", "gcode"]
 
 [tool.poetry.dependencies]
 python = ">=3.11,<3.13"

tests/test_find.py

@@ -1,4 +1,34 @@
+from cnceye.config import MYSQL_CONFIG
 from cnceye.edge import find
+import mysql.connector
+import sqlite3
+
+
+def copy_sqlite_db_to_mysql():
+    db_path = "tests/fixtures/db/listener.db"
+    conn = sqlite3.connect(db_path)
+    cur = conn.cursor()
+    query = "SELECT x,y,z,distance FROM coord"
+    data = []
+    for row in cur.execute(query):
+        data.append(row)
+    cur.close()
+    conn.close()
+
+    mysql_conn = mysql.connector.connect(**MYSQL_CONFIG, database="coord")
+    mysql_cur = mysql_conn.cursor()
+    mysql_cur.executemany(
+        "INSERT INTO sensor(x, y, z, distance) VALUES (%s, %s, %s, %s)", data
+    )
+    mysql_conn.commit()
+    mysql_cur.close()
+    mysql_conn.close()
+
+
+def test_find_edges():
+    copy_sqlite_db_to_mysql()
+    measured_edges = find.find_edges()
+    assert len(measured_edges) == 16
 
 
 def test_find_edge():
@@ -15,8 +45,7 @@ def test_find_edge_from_sqlite():
 
 
 def test_add_measured_edge_coord():
-    db_path = "tests/fixtures/db/listener.db"
-    measured_edges = find.find_edges_from_sqlite(db_path, 100.0)
+    measured_edges = find.find_edges()
     edge_data = [
         (1, 6, -50.0, 0.0, 10.0),
         (2, 8, -25.0, 38.0, 10.0),