智能水表项目 - 记一次完整的修改日志

一、前言

写给亲爱的师弟鹿同学，这是一份关于增加梅花针和指针精搜索功能的开发日志文档，希望你尽快接手这个项目，要不然师兄的论文真的就来不及啦！

二、准备工作

2.1 明确需求

在做动笔改代码之前，第一件事情就是明确我们的需求是什么，我近期简单整理了一下需要修改的内容，放到了 README.md中，这个文件相当于github的欢迎页，打开网站就能看到。

今天我们主要完成梅花针和指针的精定位功能。

2.2 梳理思路

在原有的代码中，前端web页面标记了梅花针和指针的大致位置之后，将相关的位置坐标写入到了我们的配置文件中，如:

WaterMeter:
  Choose: DN15
  DN15:
    code: '19100304122008003284'
    code_x1: 934
    code_x2: 1188
    code_y1: 215
    code_y2: 495
    completed: true
    meihua_count: 10
    meihua_r: 86
    meihua_x: 917
    meihua_y: 629
    pointer_r: 104
    pointer_x: 805
    pointer_y: 448

这里面code_x1,code_x2表示二维码的左上和右下两个角点，meihua_x meihua_y则分别代表了可视化标记时的圆心、半径等信息。因此，我们的需要读取这部分的值和图像，然后进行圆搜索或矩形搜索，再将值写回配置文件中。

那么接下来我们需要提纲挈领的思考一下整体的设计：

Q: 在哪执行这个操作?

很明显我们应该在程序初始化时执行一次这样的操作。后续我们会和甲方沟通，在执行一次测量之前，先执行一次初始化(init)，再开始(start)。接下来我将展示如何以API层为突破口看懂整个程序的。
既然是在初始化部分添加内容，当然不能像无头苍蝇一样在整个程序搜索init。从业务逻辑或者说人的直觉也好，都应该从API层作为入口进行思考。阅读app/web/api_handlers搜索init，或者看我之前写的接口文档也好，不难找到执行init的代码:

@router.post("/api/init")
async def init_measurement():
    """
    整体初始化接口。
    调用 measure_service 的 init 函数进行初始化。
    """
    from app.services.measure_service import measure_service
    try:
        # 调用 measure_service 的 init 函数进行初始化
        measure_service.init()
        return JSONResponse(content={
            "status": "success",
            "message": "测量服务初始化成功"
        })
    except ValueError as e:
        raise HTTPException(status_code=400, detail=f"初始化失败: {e}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"初始化失败: {str(e)}")

这里需要稍微一点点Fastapi、装饰器、异步、路由、try语法的知识，不过不用太探究，你只需要了解到：函数init_measurement绑定到了api/init上，当外部执行POST http:// { 你的ip }:8000/api/init时，会调用这个函数。接下来，使用了tryu - except的语法，如果执行init失败，则抛出异常，返回给用户。

可以看到，主要是执行了一个measure_service.init()函数。按住ctrl点击这个函数，即可跳转。他在/app/services/measure_service中。

    def init(self):
        """
        初始化测量服务。
        根据选择的水表产品，识别ROI内的二维码、更新梅花针和指针的监测点位置。
        刷新一次相机焦距和光源亮度参数。
        考虑该方法在每次选择水表产品后调用一次。
        """
        try:
            self.refresh()
            self.init_code()
            if 'linux' in platform.platform():
                # 在windows调试时无设备，跳过光源和相机刷新
                syscontrol_service.refresh_focus()
                syscontrol_service.refresh_light_intensity()
            param_service.set("system.status", "idle", True)
        except Exception as e:
            logger.error(f"初始化测量服务失败: {e}")
            param_service.set("system.status", "error", True)
            raise e

简单分析一下这个函数，refresh主要是刷新了一下和产品相关的参数，然后把config中的参数传给了检测器。后面init_code是检测二维码，以及光源和焦距的控制部分，不过这部分不是我们现在需要关心的。

    def refresh(self):
        """
        刷新参数。读入选中产品的参数信息。
        """
        self.params = {}
        self.product = param_service.get("WaterMeter.Choose")

        if not self.product:
            raise ValueError("未选择水表产品，无法刷新测量服务参数。")
        param_names = [
            'code_x1', 'code_x2', 'code_y1', 'code_y2',
            'meihua_r', 'meihua_x', 'meihua_y', 'meihua_count',
            'pointer_r', 'pointer_x', 'pointer_y'
        ]
        # 更新参数
        for param_name in param_names:
            self.params[param_name] = param_service.get(f"WaterMeter.{self.product}.{param_name}")
        # 检查所有参数是否都存在
        if any(value is None for value in self.params.values()):
            raise ValueError(f"产品 '{self.product}' 的参数不完整，无法刷新测量服务参数!")
        # 更新梅花针监测点位置
        self.meihua_detector = MeihuaDetector(x=self.params['meihua_x'] + 60, y=self.params['meihua_y'])
        # 更新指针监测点位置
        self.pointer_detector = PointerAngleDetector(x=self.params['pointer_x'], y=self.params['pointer_y'])

分析代码，我们现在需要进行圆搜索的代码需要添加在这个refresh之前。

Q: 需要做什么?

1. 我们首先需要写一个圆检测器，接收圆心、半径等信息，然后对图像进行圆搜索，把搜索到的圆心和半径返回。

2. 再在measure.service写一个init_search，负责调用刚刚写的圆搜索器。首先从参数服务中读取对应的值，然后获取到精搜索之后的值，再写回参数中。

3. 最后，我们在api_handler中写一个接口，用于单独测试这个功能是否正常，然后看你心情写到前端中。（前端可以交给AI）

可能看起来有点乱，为什么要这么写?
在软件工程中，需要遵循“高内聚、低耦合”的特点，以实现功能的复用。因此最基础的圆搜索器，他更像是一个工具箱，不参与整个业务逻辑。无论你是用搜索器搜索梅花针还是指针，可以调用两次圆搜索器复用代码。之后可能还需要搜索整个水表的位置，进行位置修正，搜索水表表盘这个大圆显然是也需要调用圆搜索器的，如果设计之初就参杂了一些业务逻辑，在之后复用的时候会很麻烦。

Q: 为什么只搜索圆?

我尝试过矩形搜索，因为我们的矩形搜索是搜索二维码的位置，然后裁剪区域送到二维码搜索器，供其检测。但是矩形搜索对二维码和条码的识别效果特别差。于是我想的办法是，用二维码搜索结果解析出的角点值，取xy最小为左上角点，xy最大为左下角点，然后写入配置文件中。这一部分在measure_service的init_code中。缺点是二维码识别的精度特别特别差，这个需要你以后进行改进，比如说用二值化、形态学处理等算法识别二维码。

三、圆搜索器

我在app/algorithm/detector中创建了一个新的文件searcher.py用于实现圆搜索器。然后把我们的需求以注释的形式补充。

import cv2
import numpy as np

class ElementSearcher:
    """
    该模块用于搜索水表图像中的特定元素，进行精定位。
    Methods:
    - circle_search: 搜索圆形元素
    """
    def __init__(self) -> None:
        pass

    def circle_search(self, x,y,r, image):
        """
        首先根据输入的监测点位置和半径，在图像中截取感兴趣区域。
        然后使用霍夫变换搜索图像中的圆形元素。
        由于ROI区域应略大于搜索到的圆，因此只返回搜索到的最大的圆。
        参数:
        - x: 监测点中心x
        - y: 监测点中心y
        - r: 检测区域半径

        返回:
        - circles: 检测到的圆坐标，圆由(x, y, radius)表示
        """
        pass

接下来就是AI大显身手的时候了。参考提示词:

完善circle_search，实现其功能，并给我一个测试示例__main__，读取本地文件output.mp4，测试效果，代码结构如下:...

具体代码内容参照修改后的就好，我就不进行赘述了。经过测试，发现他能搜索到红色指针的表盘圆，但是无法识别得到梅花针的圆。

显然梅花针并不是一个规则的圆形，看来我们只能是想想别的办法了。

由于梅花针呈中心对称的特性，因此我们可以写一个质心搜索器。

四、质心搜索器

所谓质心搜索器，简而言之就是去找算平均值，即梅花针区域的黑色像素平均坐标。思路有了，我们就可以接着动笔了：

def centroid_search(self, x, y, r, image):
    """
    首先根据输入的监测点位置和半径，在图像中截取感兴趣区域。
    搜索对象为黑色，先进行二值化提取黑色区域，
    然后使用质心搜索计算中心和半径。
    参数:
    - x: 监测点中心x
    - y: 监测点中心y
    - r: 检测区域半径
    - image: 输入图像

    返回:
    - circles: 检测到的质心坐标，并由(x, y, radius)表示
    """
    pass

嗯，非常完美接下来就可以写服务的部分了。（其实也没有那么完美，AI给的质心搜索部分的代码半径算的不对，让他改了几次才改好）

五、初始化搜索服务

在/app/services/measure_service中新建一个函数，init_search，为了后面异常返回和方便调试，我们先把函数的框架搭好：

def init_search(self):
	"""
	精搜索初始化接口。
	在初始化时调用一次，更新梅花针和指针的监测点位置。
	成功后写入后台参数并更新。
	"""
	try:
		pass
	except Exception as e:
		logger.error(f"初始化精搜索失败: {e}")
		raise e

在这一部分，就要引入日志系统和异常返回，因此程序的框架必须是try-except结构。关于日志部分，可以参考我发现很多程序员都不会打日志 哔哩哔哩_bilibili
接着，我们参照刚刚的给出的调用示例，完善这个函数。

六、API调用

刚刚写好了精搜索的服务，但是还没有办法测试他好不好用，因此我们还需要写一个api调用测试。在/app/web/api_handlers中添加一个:

@router.post("/api/init/search")
async def init_search():
    """
    精搜索初始化接口。
    """
    from app.services.measure_service import measure_service
    try:
        measure_service.init_search()
        return JSONResponse(content={"status": "success", "message": "精搜索初始化成功"})
    except ValueError as e:
        raise HTTPException(status_code=400, detail=f"精搜索初始化失败: {e}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"服务器内部错误: {e}")

没什么好说的，把之前的复制一下，换一下try肚子里的函数就好了。

七、调试

好了，万事俱备，让我们打开apipost调试，注意使用POST，这里多提一嘴，POST比GET安全，一般获取参数的时候使用GET，修改参数或者控制系统用POST，符合规范。

没什么问题，接下来把这部分加到init中。其实也有点BUG，填错了参数，稍微改了一下
由于在init_search的时候refresh了，因此删掉init的refresh取而代之就行。
最后，我们完整的测试标记 -> 初始化 这个流程，并测试乱标记能否正常报错返回提示

有点BUG，圆搜索的阈值需要调整一下，参照：OpenCV-Python教程：霍夫变换~圆形(HoughCircles)cv2.houghcircles-CSDN博客

测试乱标记能否正常报错：

好，没什么问题了，这个功能就算是添加完成了。

八、结语

8.1 单例

可能你对线程和进程的概念比较懵，不知道在哪里执行的init，这里我给你大概展示一下：
首先我们在measure.service的文件结尾有一句:

measure_service = MeasureService()

这个写法叫单例。整个测量服务就变成了全局变量（全局对象）。参考Python单例模式详解：从原理到实战的完整指南_python 单例-CSDN博客
在其他文件调用时，就调用这个实例化后的对象即可。比如：

from app.services.measure_service import measure_service

因此我们刚刚写的函数是web线程中运行的。事实上这种初始化部分的图像处理，对于实时性要求
并不高，因此没必要把这部分放到处理线程中。

8.2 绘图

调试的时候我是没有单独动过绘制部分的，是不过我认为有必要介绍一下图像绘制。这部分在app\algorithm\process_thread.py的draw_overlay部分，他是直接读取config.yaml的参数来进行绘制的。

8.3 上传

最后的最后，我们要把今天的工作上传到github，以便同步大家的进度。我这里再以VScode为例演示：
首先要填写提交的message，这部分可以参考约定式提交，不过我们没什么讲究，大家能看懂就行。

---

大功告成！