Prometheus告警带图完美解决方案

news2024/11/27 7:18:38

需求背景

告警分析处理流程

通常我们收到 Prometheus 告警事件通知后,往往都需要登录 Alertmanager 页面查看当前激活的告警,如果需要分析告警历史数据信息,还需要登录 Prometheus 页面的在 Alerts 中查询告警 promQL 表达式,然后复制到 Graph 中查询数据。

这样做无疑大大降低了故障分析和处理流程,此时有些聪明的小伙伴就在想,能不能在告警事件推送的时候,将这些历史数据绘制成图表一块推送过来,方便我们第一时间了解指标变化趋势,便于更快的处理问题。

功能需求分析

  1. 可靠性要求:无论告警图表数据是否能获取到,都不能影响告警信息的及时推送。(这是 webhook 最核心的功能,告警带图只是锦上添花,不可本末倒置,不能因为获取图表失败而导致告警事件无法正常推送。)
  2. 数据折线图显示:希望可以查询指定时间范围内数据变化趋势,便于分析排查问题,这是最核心的功能。
  3. 只显示异常 instance 数据:每个 job 都会对应很多 targets,我们只希望显示异常 instance 相关的数据,避免整个图表因数据过多导致显示混乱。
  4. 最大值最小值最近值显示:这三个值是我们最关注的信息,尤其是最近值,在告警触发时还是告警恢复后,我们都需要知道这个指标当前值是多少。
  5. 时间范围灵活:有些告警比较灵敏,for 时间会较短,通常不到 1 分钟(例如网络探针检测 http 状态码)。而有些告警比较迟钝,for 时间会设置的比较长,通常以为小时为单位(例如证书有效期监控)。而此时告警图表时间范围不应该统一设置为某个指定的值,而是根据 for 配置的时间灵活变化。

实现思路与效果

实现效果

废话不多说,先展示一下最终的告警效果。

在这里插入图片描述

grafana 渲染指标图:

在这里插入图片描述

方案对比分析

针对这个需求,目前主流的实现方案主要有以下两种。

方案 1:开发程序,查询 prometheus 指标数据,然后使用图表库渲染出图片,以 python 为例,可以使用 Matplotlib 或 Pandas 实现 。

优点:实现起来较为灵活,可以很好的满足各种功能需求。

缺点:图表样式单一,如果需要修改图表样式需要变更代码。

方案 2:使用 grafana-image-renderer插件渲染图片。

优点:图表样式美观,图表配置灵活方便。

缺点:需要指定 dashboard 和 panel 的 id 才能渲染,也就意味着每个告警规则都需要创建一个图表,工作量巨大。

实现思路

既然两种方案都各有优缺点,能否将两种方案的优点整合起来,既要配置实现起来灵活方便又要图表样式美观且易于配置呢?毕竟只有小孩子才做选择题,大人都是全都要。

在这里插入图片描述

具体实现思路如下:

  1. 收到 alertmanager 推送告警数据
  2. 根据告警数据,提取告警标题和异常的 instance 。
  3. 根据告警标题请求 prometheus 的 rules 接口 ,获取 for 和表达式配置。接口文档:https://prometheus.ac.cn/docs/prometheus/latest/querying/api/#rules
  4. 正则处理表达式,去除条件判断值,并新增 instance 标签选择,例如原始的告警规则表达式为,处理后的查询语句为
  5. 请求 grafana 的 api 接口,获取 grafana dashboard 配置。接口文档:https://grafana.com/docs/grafana/latest/developers/http_api/dashboard/
  6. 根据获取到的 dashboard 配置数据,替换表达式,时间范围, 告警标题。
  7. 再次 post 请求 grafana 的 api 接口,更新 dashboard 配置。
  8. 调用grafana-image-renderer插件,传入 dashboard 和 panels 参数,渲染图片并下载到本地。
  9. 将本地图片推送至企业微信、钉钉或 teams(也可上传至公有云对象存储,直接返回图片公网 url 地址)。

流程图如下:

在这里插入图片描述

核心代码与配置

grafana 配置

  1. 创建 api key 用于调用 api 接口请求 grafana。

在这里插入图片描述

  1. 创建 dashboard 和图表,并配置图表样式。

在这里插入图片描述

表达式、标题、时间范围可以随便写,主要是为了调试图表配置。

  1. 安装grafana-image-renderer 插件。

参考文档:https://grafana.com/grafana/plugins/grafana-image-renderer/。安装完成后点击渲染图像验证是否可以正常渲染。

在这里插入图片描述

webhook 程序

核心代码如下,需要注意的是为了确保可以正常获取图片和推送消息,建议添加重试和异常处理逻辑,增加程序可靠性。

import base64
import hashlib
import json
import re
import time
import os
import httpx
from log import logger
from config import vx_conf, grafana_conf, alertmanager_conf, prometheus_conf, aliyun_conf
from oss2 import Auth, Bucket, exceptions


class VxRobot:
    """
    企业微信推送
    """

    def __init__(self, team, max_retries=3, backoff_factor=2):
        self.url = vx_conf[team]
        self.headers = {'Content-Type': 'application/json'}
        self.params = {}
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor

    def __send_data(self):
        """
        推送企业微信数据
        :return:
        """
        retries = 0
        with httpx.Client(verify=False) as client:
            while retries < self.max_retries:
                try:
                    # 发送 POST 请求
                    response = client.post(self.url, headers=self.headers, json=self.params, timeout=10)
                    response.raise_for_status()  # 如果状态码不是 2xx,会抛出异常
                    if response.json()['errcode'] == 0:
                        logger.info("企业微信告警推送完成")
                        return
                    else:
                        logger.error(f"企业微信告警推送内容有误: {response.json()['errmsg']}")
                        break
                except (httpx.RequestError, httpx.HTTPStatusError) as exc:
                    retries += 1
                    logger.error(f"企业微信告警推送失败 (尝试 {retries}/{self.max_retries}): {exc}")
                    if retries < self.max_retries:
                        sleep_time = self.backoff_factor * retries
                        logger.info(f"等待 {sleep_time} 秒后重试...")
                        time.sleep(sleep_time)  # 实现退避策略
                    else:
                        logger.error("达到最大重试次数,企业微信告警推送失败")
                        raise Exception(f"无法使用企业微信告警推送: {exc}")

    def send_img(self, image_path):
        """
        推送图片信息
        :param image_path:图片路径
        :return: None
        """
        try:
            # 读取图片文件内容
            with open(image_path, "rb") as f:
                image_data = f.read()
            # 计算图片内容的 MD5 值
            md5_hash = hashlib.md5(image_data).hexdigest()
            # 生成图片内容的 Base64 编码
            base64_encoded = base64.b64encode(image_data).decode("utf-8")
        except FileNotFoundError:
            raise FileNotFoundError(f"图片文件 {image_path} 未找到")
        except Exception as e:
            raise RuntimeError(f"处理图片时发生错误: {str(e)}")
        params = {
            "msgtype": "image",
            "image": {
                "base64": "DATA",
                "md5": "MD5"
            }
        }
        # 数据替换
        params["image"]["base64"] = base64_encoded
        params["image"]["md5"] = md5_hash
        logger.info("推送企业微信图片内容,图片地址为%s" % image_path)
        self.params = params
        self.__send_data()


def get_panel(alert_name, instance):
    """
    获取指标图表信息
    :return:
    """
    prometheus = PrometheusTools()
    config = prometheus.get_alert_config(alert_name)
    config['instance'] = instance
    config['time_range'] = 'now-' + str(2 * config['duration']) + 's'
    if 2 * config['duration'] > 60:
        config['alert_name'] = config['alert_name'] + '(最近' + str(round(2 * config['duration'] / 60)) + '分钟)'
    else:
        config['alert_name'] = config['alert_name'] + '(最近' + str(round(2 * config['duration'])) + '秒)'
    grafana = GrafanaTools()
    dashboard_data = grafana.fetch_dashboard()
    dashboard_conf = grafana.update_panel_query(dashboard_data, config)
    grafana.update_dashboard(dashboard_conf)
    local_img_path = grafana.render_image(config['alert_name'])
    return local_img_path
    # aliyun_oss = AliyunTools()
    # oss_path = local_img_path.replace("img/", "alert_img/")
    # aliyun_oss.upload_image(local_img_path, oss_path)


class PrometheusTools:
    def __init__(self, max_retries=3, backoff_factor=2):
        """
        初始化 Prometheus 客户端
        """
        self.url = prometheus_conf['url']
        self.username = prometheus_conf['username']
        self.password = prometheus_conf['password']
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor

        # 生成认证头
        auth_str = f"{self.username}:{self.password}"
        auth_bytes = base64.b64encode(auth_str.encode("utf-8")).decode("utf-8")
        self.headers = {"Authorization": f"Basic {auth_bytes}"}

    def get_alert_config(self, alert_name):
        """
        获取告警配置
        :param alert_name: 告警名称
        :return: 告警配置
        """
        url = f"{self.url}/api/v1/rules"
        params = {"rule_name[]": alert_name}
        retries = 0

        with httpx.Client(verify=False) as client:
            while retries < self.max_retries:
                try:
                    # 发起请求
                    response = client.get(url, params=params, headers=self.headers, timeout=10)
                    response.raise_for_status()  # 如果状态码不是 2xx,会抛出异常
                    logger.info("成功获取 Prometheus 告警数据")
                    logger.debug(json.dumps(response.json()))
                    # 提取关键数据
                    config = {
                        'alert_name': response.json()['data']['groups'][0]['rules'][0]['name'],
                        'query': response.json()['data']['groups'][0]['rules'][0]['query'],
                        'duration': response.json()['data']['groups'][0]['rules'][0]['duration']
                    }
                    logger.debug(config)
                    return config
                except (httpx.RequestError, httpx.HTTPStatusError) as exc:
                    logger.error(f"Attempt {retries + 1} failed: {exc}")
                    retries += 1
                    if retries < self.max_retries:
                        time.sleep(self.backoff_factor * retries)  # 退避策略
                    else:
                        raise Exception(f"Failed to fetch alert config after {self.max_retries} retries.")


class GrafanaTools:
    """
    grafana渲染图表工具
    """

    def __init__(self, max_retries=3, backoff_factor=2):
        self.grafana_url = grafana_conf['url']
        self.api_key = grafana_conf['api_key']
        self.dashboard_uid = grafana_conf['dashboard_id']
        self.panel_id = grafana_conf['panel_id']
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor

    def fetch_dashboard(self):
        """
        获取仪表盘的完整配置
        :return: 仪表盘的完整配置
        """
        url = f"{self.grafana_url}/api/dashboards/uid/{self.dashboard_uid}"
        logger.info("开始获取dashboard配置, url: %s" % url)

        retries = 0
        with httpx.Client(verify=False) as client:
            while retries < self.max_retries:
                try:
                    # 发送请求
                    response = client.get(url, headers=self.headers, timeout=10)
                    response.raise_for_status()  # 如果状态码不是 2xx,会抛出异常
                    logger.info("成功获取dashboard配置")
                    logger.debug(json.dumps(response.json(), indent=2))  # 打印详细配置
                    return response.json()
                except (httpx.RequestError, httpx.HTTPStatusError) as exc:
                    retries += 1
                    logger.error(f"请求失败 (尝试 {retries}/{self.max_retries}): {exc}")
                    if retries < self.max_retries:
                        sleep_time = self.backoff_factor * retries
                        logger.info(f"等待 {sleep_time} 秒后重试...")
                        time.sleep(sleep_time)  # 实现退避策略
                    else:
                        logger.error("达到最大重试次数,获取dashboard配置失败")
                        raise Exception(f"无法获取仪表盘配置: {exc}")

    def update_panel_query(self, dashboard_data, config):
        """
        修改grafana dashboard配置内容
        :param dashboard_data: dashboard配置
        :param config: 告警配置
        :return: 新的grafana dashboard配置
        """
        # logger.error(config)
        for panel in dashboard_data["dashboard"]["panels"]:
            if panel["id"] == self.panel_id:
                # 修改面板查询,动态添加 instance 变量
                panel["title"] = config['alert_name']
                instance = config['instance']
                query = config['query']
                # 去除规则条件
                query_prom = re.sub(r'\s*(==|!=|>=|<=|>|<)\s*([0-9]+(?:\.[0-9]+)?)$', '', query)
                # 正则表达式,匹配 Prometheus 查询语句中的指标名称
                pattern = r'([a-z_][a-z0-9_]*)(?=\s*(\{|\[|$))'
                # 查找匹配的指标名称
                match = re.search(pattern, query_prom)
                # 如果找到了匹配项
                if match and instance != "none":
                    # 获取指标名称的结束位置
                    end_pos = match.end(1)  # end(1) 获取捕获组1(指标名称)的结束位置
                    # 判断原始指标是否存在标签选择
                    if '{' in query_prom:
                        query_cleaned = query_prom[:end_pos + 1] + 'instance="' + instance + '",' + query_prom[
                                                                                                    end_pos + 1:]
                    else:
                        query_cleaned = query_prom[:end_pos] + '{instance="' + instance + '"}' + query_prom[end_pos:]
                else:
                    logger.error("指标匹配失败,使用默认指标%s" % query_prom)
                    query_cleaned = query_prom
                # logger.error(query_cleaned)
                panel["targets"][0]["expr"] = query_cleaned
                break
        logger.info("dashboard配置替换完成")
        logger.debug(json.dumps(dashboard_data))
        return dashboard_data

    def update_dashboard(self, dashboard):
        """
        更新仪表盘配置到 Grafana
        :param dashboard: 要更新的仪表盘配置 (dict)
        :return: None
        """
        url = f"{self.grafana_url}/api/dashboards/db"
        logger.info("开始更新Dashboard配置, url: %s" % url)

        retries = 0
        with httpx.Client(verify=False) as client:
            while retries < self.max_retries:
                try:
                    # 发送 POST 请求
                    response = client.post(url, headers=self.headers, json=dashboard, timeout=10)
                    response.raise_for_status()  # 如果状态码不是 2xx,会抛出异常
                    logger.info("Dashboard更新完成")
                    return
                except (httpx.RequestError, httpx.HTTPStatusError) as exc:
                    retries += 1
                    logger.error(f"更新Dashboard失败 (尝试 {retries}/{self.max_retries}): {exc}")
                    if retries < self.max_retries:
                        sleep_time = self.backoff_factor * retries
                        logger.info(f"等待 {sleep_time} 秒后重试...")
                        time.sleep(sleep_time)  # 实现退避策略
                    else:
                        logger.error("达到最大重试次数,Dashboard更新失败")
                        raise Exception(f"无法更新仪表盘配置: {exc}")

    def render_image(self, alert_name):
        """
        调用 Grafana 渲染图片 API
        :param alert_name: 告警标题
        :return: 图片保存路径
        """
        output_path = "img/" + alert_name + ".png"
        render_url = (
            f"{self.grafana_url}/render/d-solo/{self.dashboard_uid}"
            f"?orgId=1&panelId={self.panel_id}&width=2000&height=1000"
        )
        logger.info("开始渲染图片, render_url: %s" % render_url)

        retries = 0
        with httpx.Client(verify=False) as client:
            while retries < self.max_retries:
                try:
                    # 发起请求
                    response = client.get(render_url, headers=self.headers, timeout=20)
                    response.raise_for_status()  # 如果状态码不是 2xx,会抛出异常
                    # 确保输出目录存在
                    os.makedirs(os.path.dirname(output_path), exist_ok=True)
                    with open(output_path, "wb") as f:
                        f.write(response.content)
                    logger.info(f"告警图片下载至 {output_path}")
                    return output_path
                except (httpx.RequestError, httpx.HTTPStatusError) as exc:
                    retries += 1
                    logger.error(f"渲染图片失败 (尝试 {retries}/{self.max_retries}): {exc}")
                    if retries < self.max_retries:
                        sleep_time = self.backoff_factor * retries
                        logger.info(f"等待 {sleep_time} 秒后重试...")
                        time.sleep(sleep_time)  # 实现退避策略
                    else:
                        logger.error("达到最大重试次数,渲染图片失败")
                        raise Exception(f"无法渲染图片: {exc}")


class AliyunTools:
    def __init__(self, max_retries=3, backoff_factor=2):
        self.auth = Auth(aliyun_conf['access_key_id'], aliyun_conf['access_key_secret'])
        self.bucket = Bucket(self.auth, aliyun_conf['endpoint'], aliyun_conf['bucket_name'])
        self.bucket_name = aliyun_conf['bucket_name']
        self.endpoint = aliyun_conf['endpoint']
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor

    def upload_image(self, local_image_path, oss_object_name):
        """
        上传图片到阿里云 OSS
        :param local_image_path: 本地图片路径
        :param oss_object_name: OSS 中的对象名 (路径/文件名)
        :return: 图片的公开访问地址
        """
        if not os.path.exists(local_image_path):
            raise FileNotFoundError(f"本地文件 {local_image_path} 不存在")

        retries = 0
        while retries < self.max_retries:
            try:
                logger.info(f"开始上传图片 {local_image_path} 到 OSS,目标对象名: {oss_object_name}")
                # 上传图片到 OSS
                self.bucket.put_object_from_file(oss_object_name, local_image_path)

                # 获取图片访问地址
                image_url = self.bucket.sign_url('GET', oss_object_name, 60 * 60 * 24)
                logger.info(f"图片成功上传至 OSS,访问地址: {image_url}")
                return image_url
            except (exceptions.RequestError, exceptions.ServerError) as exc:
                retries += 1
                logger.error(f"上传图片到 OSS 失败 (尝试 {retries}/{self.max_retries}): {exc}")
                if retries < self.max_retries:
                    sleep_time = self.backoff_factor * retries
                    logger.info(f"等待 {sleep_time} 秒后重试...")
                    time.sleep(sleep_time)
                else:
                    logger.error("达到最大重试次数,上传图片失败")
                    raise Exception(f"无法上传图片到 OSS: {exc}")

查看更多

崔亮的博客-专注devops自动化运维,传播优秀it运维技术文章。更多原创运维开发相关文章,欢迎访问https://www.cuiliangblog.cn

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/2248275.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

深入理解 Java 基本语法之运算符

&#xff08;一&#xff09;研究背景 在 Java 编程中&#xff0c;运算符是处理数据和变量的基本工具&#xff0c;掌握各种运算符的使用方法对于提高编程效率至关重要。 &#xff08;二&#xff09;研究目的 深入理解 Java 基础运算符的概念、分类和作用&#xff0c;通过具体…

iOS 17.4 Not Installed

0x00 系统警告 没有安装 17.4 的模拟器&#xff0c;任何操作都无法进行&#xff01; 点击 OK 去下载&#xff0c;完成之后&#xff0c;依旧是原样&#xff01; 0x01 解决办法 1、先去官网下载对应的模拟器&#xff1a; https://developer.apple.com/download/all/?q17.4 …

Flink细粒度的资源管理

Apache Flink致力于为所有应用程序自动导出合理的默认资源需求。对于希望根据其特定场景微调其资源消耗的用户,Flink提供了细粒度的资源管理。这里我们就来看下细粒度的资源管理如何使用。(注意该功能目前仅对DataStream API有用) 1. 适用场景 使用细粒度的资源管理的可能…

Ubuntu20.04运行msckf_vio

文章目录 环境配置修改编译项目运行MSCKF_VIO运行 Launch 文件运行 rviz播放 ROSBAG 数据集 运行结果修改mskcf 保存轨迹EVO轨迹评价EVO轨迹评估流程实操先把euroc的真值转换为tum&#xff0c;保存为data.tum正式评估 报错1问题描述 报错2问题描述问题分析问题解决 参考 环境配…

计算机网络 第4章 网络层

计算机网络 &#xff08;第八版&#xff09;谢希仁 第 4 章 网络层4.2.2 IP地址**无分类编址CIDR**IP地址的特点 4.2.3 IP地址与MAC地址4.2.4 ARP 地址解析协议4.2.5 IP数据报的格式题目2&#xff1a;IP数据报分片与重组题目&#xff1a;计算IP数据报的首部校验和(不正确未改) …

Angular面试题汇总系列一

1. 如何理解Angular Signal Angular Signals is a system that granularly tracks how and where your state is used throughout an application, allowing the framework to optimize rendering updates. 什么是信号 信号是一个值的包装器&#xff0c;可以在该值发生变化时…

SAR ADC系列15:基于Vcm-Base的开关切换策略

VCM-Based开关切换策略&#xff1a;采样~第一次比较 简单说明: 电容上下极板分别接Vcm&#xff08;一般Vcm1/2Vref&#xff09;。采样断开瞬间电荷锁定&#xff0c;进行第一次比较。 当VIP > VIN 时&#xff0c;同时 减小VIP 并 增大VIN 。P阵列最高权重电容从Vcm(1/2Vref)…

实现Excel文件和其他文件导出为压缩包,并导入

导出 后端&#xff1a; PostMapping("/exportExcelData")public void exportExcelData(HttpServletRequest request, HttpServletResponse response, RequestBody ResData resData) throws IOException {List<Long> menuIds resData.getMenuIds();List<Co…

某车企ASW面试笔试题

01--背景 去年由于工作岗位的动荡&#xff0c;于是面试了一家知名车企&#xff0c;上来进行了一番简单的介绍之后&#xff0c;被告知需要进入笔试环节&#xff0c;以往单位面试都是简单聊聊技术问题&#xff0c;比如对软件开发的流程或者使用的工具等待问题的交流&#xff0c;…

计算(a+b)/c的值

计算&#xff08;ab&#xff09;/c的值 C语言代码C语言代码Java语言代码Python语言代码 &#x1f490;The Begin&#x1f490;点点关注&#xff0c;收藏不迷路&#x1f490; 给定3个整数a、b、c&#xff0c;计算表达式(ab)/c的值&#xff0c;/是整除运算。 输入 输入仅一行&…

【在Linux世界中追寻伟大的One Piece】多线程(二)

目录 1 -> 分离线程 2 -> Linux线程互斥 2.1 -> 进程线程间的互斥相关背景概念 2.2 -> 互斥量mutex 2.3 -> 互斥量的接口 2.4 -> 互斥量实现原理探究 3 -> 可重入VS线程安全 3.1 -> 概念 3.2 -> 常见的线程不安全的情况 3.3 -> 常见的…

【NLP高频面题 - 分布式训练】ZeRO1、ZeRO2、ZeRO3分别做了哪些优化?

【NLP高频面题 - 分布式训练】ZeRO1、ZeRO2、ZeRO3分别做了哪些优化&#xff1f; 重要性&#xff1a;★★ NLP Github 项目&#xff1a; NLP 项目实践&#xff1a;fasterai/nlp-project-practice 介绍&#xff1a;该仓库围绕着 NLP 任务模型的设计、训练、优化、部署和应用&am…

AIGC--AIGC与人机协作:新的创作模式

AIGC与人机协作&#xff1a;新的创作模式 引言 人工智能生成内容&#xff08;AIGC&#xff09;正在以惊人的速度渗透到创作的各个领域。从生成文本、音乐、到图像和视频&#xff0c;AIGC使得创作过程变得更加快捷和高效。然而&#xff0c;AIGC并非完全取代了人类的创作角色&am…

C++11特性(详解)

目录 1.C11简介 2.列表初始化 3.声明 1.auto 2.decltype 3.nullptr 4.范围for循环 5.智能指针 6.STL的一些变化 7.右值引用和移动语义 1.左值引用和右值引用 2.左值引用和右值引用的比较 3.右值引用的使用场景和意义 4.右值引用引用左值及其一些更深入的使用场景分…

React中事件处理和合成事件:理解与使用

&#x1f90d; 前端开发工程师、技术日更博主、已过CET6 &#x1f368; 阿珊和她的猫_CSDN博客专家、23年度博客之星前端领域TOP1 &#x1f560; 牛客高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》 &#x1f35a; 蓝桥云课签约作者、上架课程《Vue.js 和 E…

大数据新视界 -- 大数据大厂之 Hive 数据桶:优化聚合查询的有效手段(下)(10/ 30)

&#x1f496;&#x1f496;&#x1f496;亲爱的朋友们&#xff0c;热烈欢迎你们来到 青云交的博客&#xff01;能与你们在此邂逅&#xff0c;我满心欢喜&#xff0c;深感无比荣幸。在这个瞬息万变的时代&#xff0c;我们每个人都在苦苦追寻一处能让心灵安然栖息的港湾。而 我的…

基于FPGA的信号DM编解码实现,包含testbench和matlab对比仿真

目录 1.算法运行效果图预览 2.算法运行软件版本 3.部分核心程序 4.算法理论概述 1.编码器硬件结构 2.解码器硬件结构 5.算法完整程序工程 1.算法运行效果图预览 (完整程序运行后无水印) FPGA测试结果如下&#xff1a; matlab对比仿真结果如下&#xff1a; 2.算法运行软…

鸿蒙中拍照上传与本地图片上传

1.首页ui import { picker } from kit.CoreFileKit; import fs from ohos.file.fs; import request from ohos.request; import { promptAction } from kit.ArkUI; import { cameraCapture } from ./utils/CameraUtils; import { common } from kit.AbilityKit; import { Imag…

【算法】连通块问题(C/C++)

目录 连通块问题 解决思路 步骤&#xff1a; 初始化&#xff1a; DFS函数&#xff1a; 复杂度分析 代码实现&#xff08;C&#xff09; 题目链接&#xff1a;2060. 奶牛选美 - AcWing题库 解题思路&#xff1a; AC代码&#xff1a; 题目链接&#xff1a;687. 扫雷 -…

人工智能 实验2 jupyter notebook平台 打印出分类器的正确率

实验2 jupyter notebook平台 【实验目的】掌握jupyter notebook平台的使用方法 【实验内容】上传文件到jupyter notebook平台&#xff0c;学会编辑运行ipynb文件 【实验要求】写明实验步骤&#xff0c;必要时补充截图 安装Anaconda。 2、 将BreadCancer.zip上传到jupyter no…