Full end-to-end support.

Cleaned up a lot of code.  Some things are still kind of messy, but it's a vast improvement over what I used to have.  Data is easy to pass around the system, and the pattern info has proved invaluable for cleanly connecting prints, scans, and analysis.

.devcontainer/Dockerfile

@@ -2,3 +2,4 @@ FROM debian:bullseye-slim
 
 RUN apt-get update && apt-get install -y python3 python3-pip git ffmpeg
 RUN pip3 install opencv-python-headless matplotlib aiohttp requests
+RUN pip3 install websocket-client

.devcontainer/devcontainer.json

@@ -7,6 +7,7 @@
             "BUILDKIT_INLINE_CACHE": "0"
           }
     },
+    "runArgs": ["--device=/dev/video2"],
     "extensions": ["ms-python.python", "076923.python-image-preview"]
 
-}
+}

analysis.py

@@ -1,9 +1,11 @@
 import numpy as np
 from constants import *
+from processing import *
+from pa_result import PaResult
 
 def brightest_average(pixel_values: np.ndarray):
     brightest_pixels = np.argsort(pixel_values)[-3:]
-    line_brightest_x = FRAME_SIZE_X - np.average(brightest_pixels)
+    line_brightest_x = CROP_FRAME_SIZE_X - np.average(brightest_pixels)
     return line_brightest_x
 
 
@@ -14,7 +16,7 @@ def weighted_average(pixel_values: np.ndarray):
     if adjusted_values.max() == 0:
         # FIXME: I need an appropriate solution for what to do if there are no non-zero values.
         return 70
-    return FRAME_SIZE_X - np.average(x_values, weights=adjusted_values)
+    return CROP_FRAME_SIZE_X - np.average(x_values, weights=adjusted_values)
 
 
 def first_non_zero(pixel_values: np.ndarray):
@@ -39,3 +41,55 @@ def compute_x_value(pixel_values: np.ndarray):
     # return algorithms["count_non_zero"](pixel_values)
     # return algorithms["first_non_zero"](pixel_values)
     return algorithms["weighted_avg"](pixel_values)
+
+
+def generate_height_data_for_frame(frame: np.ndarray):
+    frame = crop_frame(frame)
+    frame = preprocess_frame(frame)
+    frame = apply_gaussian_blur(frame)
+
+    frame_height_data = np.ndarray(frame.shape[0])
+
+    for index, line in enumerate(frame):
+        # if line.max() > 0:
+        laser_x_val = compute_x_value(line)
+        frame_height_data[index] = laser_x_val
+
+    return frame_height_data
+
+
+def generate_height_data_from_video(video_file: str):
+    video = cv2.VideoCapture(video_file)
+    frame_count = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
+    height_data: np.ndarray = np.ndarray((frame_count, CROP_FRAME_SIZE_Y))
+
+    frame_index = 0
+    while video.isOpened():
+        ret, frame = video.read()
+        if not ret:
+            break
+
+        height_data[frame_index] = generate_height_data_for_frame(frame)
+        frame_index += 1
+
+    return height_data
+
+
+def compute_score_from_heightmap(height_map: np.ndarray):
+    sum_of_scores = 0
+
+    for line in height_map.transpose():
+        sum_of_scores += np.std(line)
+    return sum_of_scores
+
+
+def pa_score_from_video_file(video_file: str) -> PaResult:
+    video_height_data = generate_height_data_from_video(video_file)
+
+    # if OUTPUT_HEIGHT_MAPS:
+    #     graph_height_map(video_height_data, f"height_maps/{Path(video_file).stem}.png")
+
+    score = compute_score_from_heightmap(video_height_data)
+
+    return PaResult(video_file, video_height_data, score)
+

constants.py

@@ -1,9 +1,8 @@
 OUTPUT_GRAPH = False
 OUTPUT_FRAMES = False
-OUTPUT_HEIGHT_MAPS = True
+OUTPUT_HEIGHT_MAPS = False
 
-
-X_OFFSET = 200
-Y_OFFSET = 20
-FRAME_SIZE_X = 200
-FRAME_SIZE_Y = 60
+CROP_X_OFFSET = 200
+CROP_Y_OFFSET = 20
+CROP_FRAME_SIZE_X = 200
+CROP_FRAME_SIZE_Y = 60

klipper/gcode.py

@@ -1,9 +1,11 @@
 # import asyncio
 # import aiohttp
 import requests
+import websocket
 
 # HOST = 'fluiddpi.local'
-HOST = 'http://192.168.1.113'
+HOST = '192.168.1.113'
+WS_PORT = 7125
 GCODE_ENDPOINT = '/printer/gcode/script'
 OBJECTS_ENDPOINT = '/printer/objects/query'
 
@@ -29,29 +31,35 @@ def move_relative(x: float = None, y: float = None, z: float = None, f: float =
     """)
 
 
-async def send_gcode(gcode: str):
-    async with aiohttp.ClientSession() as session:
-        json_data = {
-            "script": gcode
-        }
-        async with session.post(HOST + GCODE_ENDPOINT, json=json_data) as resp:
-            return await resp.json()
+# async def send_gcode(gcode: str):
+#     async with aiohttp.ClientSession() as session:
+#         json_data = {
+#             "script": gcode
+#         }
+#         async with session.post(HOST + GCODE_ENDPOINT, json=json_data) as resp:
+#             return await resp.json()
 
 
 def send_gcode(gcode: str):
     json_data = {
         "script": gcode
     }
-    resp = requests.post(HOST + GCODE_ENDPOINT, json=json_data)
-    return resp.json()
+    resp = requests.post("http://" + HOST + GCODE_ENDPOINT, json=json_data, timeout=600)
+    return resp
+# ws = websocket.WebSocket()
+# ws.connect(f"ws://{HOST}:{WS_PORT}/klippysocket")
+
+# def send_gcode():
+#     ws.send()
+#     pass
 
 
 def home():
     return send_gcode('G28')
 
 
 def has_homed():
-    resp = requests.get(HOST + OBJECTS_ENDPOINT, params="toolhead")
+    resp = requests.get("http://" + HOST + OBJECTS_ENDPOINT, params="toolhead")
     result = resp.json()["result"]
     return result["status"]["toolhead"]["homed_axes"] == "xyz"
 
@@ -64,6 +72,21 @@ def do_initialization_routine():
         print("Rehoming Z")
         send_gcode("G28 Z")
 
+def query_printer_position():
+    resp = requests.get("http://" + HOST + OBJECTS_ENDPOINT, params="motion_report")
+    return resp.json()["result"]["status"]["motion_report"]["live_position"]
+
+def wait_until_printer_at_location(x = None, y = None, z = None):
+    while True:
+        position = query_printer_position()
+        if x is not None and abs(x - position[0]) > 0.01:
+            continue
+        if y is not None and abs(y - position[1]) > 0.01:
+            continue
+        if z is not None and abs(z - position[2]) > 0.01:
+            continue
+        break
+
 def main():
     do_initialization_routine()
     print("Finished initializing")

main.py

@@ -1,97 +1,118 @@
 #!/usr/bin/python3
-import cv2
-import numpy as np
-from glob import glob
-from pathlib import Path
-
 from processing import *
-from visualization import graph_height_map
-from analysis import compute_x_value
-
-
-def generate_height_data_for_frame(frame: np.ndarray):
-    frame = crop_frame(frame)
-    frame = preprocess_frame(frame)
-    frame = apply_gaussian_blur(frame)
+from analysis import pa_score_from_video_file
+from pattern_info import PatternInfo
+from record import record_pattern
+from pa_result import PaResult
+from pa import *
 
-    frame_height_data = np.ndarray(frame.shape[0])
+import klipper.gcode as g
 
-    for index, line in enumerate(frame):
-        # if line.max() > 0:
-        laser_x_val = compute_x_value(line)
-        frame_height_data[index] = laser_x_val
+import tempfile
+from pprint import pprint
 
-    return frame_height_data
 
+# This will print a calibrated + control pattern and measure the % improvement after tuning
+VALIDATE_RESULTS = True
 
-def generate_height_data_from_video(video_file: str):
-    video = cv2.VideoCapture(video_file)
-    frame_count = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
-    height_data: np.ndarray = np.ndarray((frame_count, FRAME_SIZE_Y))
-
-    frame_index = 0
-    while video.isOpened():
-        ret, frame = video.read()
-        if not ret:
-            break
-
-        height_data[frame_index] = generate_height_data_for_frame(frame)
-
-        frame_index += 1
 
-    return height_data
+PRINT_START = """
+G28
+M104 S180; preheat nozzle while waiting for build plate to get to temp
+M190 S{BUILD_PLATE_TEMPERATURE};
+QUAD_GANTRY_LEVEL
+CLEAN_NOZZLE
+G28 Z
+M109 S{HOTEND_TEMPERATURE};
+"""
 
+def generate_pa_results_for_pattern(pattern_info: PatternInfo)-> list[PaResult]:
+    results = []
 
-def compute_score_from_heightmap(height_map: np.ndarray):
-    sum_of_scores = 0
+    # Hardcoding a buffer distance of 3mm here for now.  Adjust if needed.
+    with tempfile.TemporaryDirectory("pa_videos") as dir:
+        video_files = record_pattern(pattern_info, 3, dir)
 
-    for line in height_map.transpose():
-        sum_of_scores += np.std(line)
-    return sum_of_scores
+        for video_file in video_files:
+            results.append(
+                pa_score_from_video_file(video_file)
+            )
+    return results
 
 
 def main():
-    ranking = []
-
-    # frame_score = compute_score_for_frame(laser_x_values)
-    # print(frame_index, frame_std)
-
-    # i = 0
-    for video_file in sorted(glob("sample_data2/*")):
-        # if i < 6:
-        #     i += 1
-        #     continue
-        video_height_data = generate_height_data_from_video(video_file)
-
-        if OUTPUT_HEIGHT_MAPS:
-            graph_height_map(video_height_data, f"height_maps/{Path(video_file).stem}.png")
-
-        score = compute_score_from_heightmap(video_height_data)
-        # height_data = compute_height_map(video_file)
-        # graph_height_map(height_data)
-
-        # return
-
-        # fig.suptitle(video_file)
-
-        # out = cv2.VideoWriter("out.avi", cv2.VideoWriter_fourcc('M','J','P','G'), 30, (400,400))
-
-        frame_index = 0
-
-        # video_std = []
-            # red_line = cv2.cvtColor(red_line, cv2.COLOR_GRAY2BGR)
-            # out.write(red_line)
-        # exit()
-        # out.release()
-        # print(np.std(video_std))
-        print(video_file, score)
-
-        ranking.append((video_file, score))
-        # return
-
-    print('\nSCORES\n')
-
-    [ print(x) for x in sorted(ranking, key=lambda x: x[1])]
+    calibration_pattern = PatternInfo(
+        0, 0.06,
+        30, 30,
+        10,
+        30, 4
+    )
+
+
+    # g.send_gcode(PRINT_START)
+    g.send_gcode("CLEAN_NOZZLE")
+    g.send_gcode(generate_pa_tune_gcode(calibration_pattern))
+    g.wait_until_printer_at_location(FINISHED_X, FINISHED_Y)
+    g.send_gcode("M104 S0; let the hotend cool")
+
+    results = generate_pa_results_for_pattern(calibration_pattern)
+
+    sorted_results = list(sorted(zip(results, calibration_pattern.pa_values), key=lambda x: x[0].score))
+    sorted_results = list([(x.score, y) for x, y in sorted_results])
+
+    best_pa_value = sorted_results[0][1]
+    print()
+    pprint(sorted_results)
+    print()
+    print(f"Recommended PA Value: {best_pa_value}, with a score of {sorted_results[0][0]}")
+    print()
+    g.send_gcode(f"SET_PRESSURE_ADVANCE ADVANCE={best_pa_value}")
+
+    if not VALIDATE_RESULTS:
+        return
+
+    control = PatternInfo(
+        0, 0,
+        65, 30,
+        10,
+        30, 4
+    )
+
+    calibrated = PatternInfo(
+        best_pa_value, best_pa_value,
+        100, 30,
+        10,
+        30, 4
+    )
+
+
+    gcode = f"""
+    M109 S{HOTEND_TEMPERATURE};
+    CLEAN_NOZZLE
+    """
+    gcode += generate_pa_tune_gcode(control, False)
+    gcode += generate_pa_tune_gcode(calibrated)
+    g.send_gcode(gcode)
+    g.send_gcode("M104 S0; let the hotend cool")
+    g.wait_until_printer_at_location(FINISHED_X, FINISHED_Y)
+
+    control_results = generate_pa_results_for_pattern(control)
+    calibrated_results = generate_pa_results_for_pattern(calibrated)
+
+    control_scores = list([x.score for x in control_results])
+    calibrated_scores = list([x.score for x in calibrated_results])
+
+    control_average = np.average(control_scores)
+    calibrated_average = np.average(calibrated_scores)
+
+    print("Control")
+    pprint(control_scores)
+    print("Calibrated")
+    pprint(calibrated_scores)
+    print()
+    print(f"Average Control Score: {control_average}")
+    print(f"Average Calibrated Score: {calibrated_average}")
+    print()
 
 
 if __name__=="__main__":