Merge branch '78:main' into main

This commit is contained in:
shunian11
2026-08-10 15:05:09 +08:00
committed by GitHub
12 changed files with 247 additions and 274 deletions
+7 -8
View File
@@ -29,22 +29,21 @@ jobs:
- name: Test build tooling
run: python -m unittest discover -s scripts/tests -v
- id: list
name: Get all variant list
run: |
echo "all_variants=$(python scripts/build.py --list-boards --json)" >> $GITHUB_OUTPUT
- id: select
name: Select variants based on changes
shell: bash
env:
ALL_VARIANTS: ${{ steps.list.outputs.all_variants }}
run: |
EVENT_NAME="${{ github.event_name }}"
# push 到 main 分支,编译全部变体
if [[ "$EVENT_NAME" == "push" ]]; then
echo "variants=$ALL_VARIANTS" >> $GITHUB_OUTPUT
# Keep the full matrix out of an environment variable. A single
# Linux environment entry is limited to 128 KiB, and the variant
# JSON can exceed that as boards are added.
{
printf 'variants='
python scripts/build.py --list-boards --json
} >> "$GITHUB_OUTPUT"
exit 0
fi
-1
View File
@@ -20,7 +20,6 @@ RUN chmod 0755 \
docker/firmware-builder/entrypoint.sh \
docker/firmware-builder/firmware_builder.py \
&& . "${IDF_PATH}/export.sh" >/dev/null \
&& python3 -m pip install --no-cache-dir oss2==2.19.1 \
&& idf.py --version \
&& python3 scripts/build.py --list-boards --json >/tmp/xiaozhi-boards.json \
&& python3 scripts/build.py --list-languages --json >/tmp/xiaozhi-languages.json
+13 -15
View File
@@ -53,28 +53,26 @@ Each successful job writes:
- `manifest.json`: inputs, tool versions, source revision, sizes, and SHA-256
checksums.
To upload the job output to OSS, also pass:
To upload the job output to an HTTP artifact receiver, also pass:
```text
FIRMWARE_OSS_UPLOAD=true
FIRMWARE_OSS_ENDPOINT=oss-cn-shenzhen.aliyuncs.com
FIRMWARE_OSS_PREFIX=custom_firmwares
OSS_BUCKET_NAME=<bucket>
OSS_ACCESS_KEY_ID=<access-key-id>
OSS_ACCESS_KEY_SECRET=<access-key-secret>
FIRMWARE_UPLOAD_URL=https://example.com/api/firmware-builds
FIRMWARE_UPLOAD_TOKEN=<upload-token>
FIRMWARE_JOB_ID=<unique-safe-job-id>
```
The builder uploads the two firmware images, `build.log`, and `manifest.json`
to `custom_firmwares/<job-id>/`. The manifest is uploaded last so consumers do
not observe a completed job before its other objects are available. Transient
OSS connection, timeout, throttling, and server errors are retried up to four
times with exponential backoff; authentication and other permanent errors fail
immediately.
The builder sends an authenticated HTTP `PUT` for the two firmware images,
`build.log`, and `manifest.json` to
`<upload-url>/<job-id>/artifacts/<filename>`. The manifest is uploaded last so
consumers do not observe a completed job before its other objects are available.
Transient connection, timeout, throttling, and server errors are retried up to
four times with exponential backoff; authentication and other permanent errors
fail immediately. Storage credentials and provider details remain entirely on
the receiving service.
Use a unique empty output directory for each job. In ECI, pass the same inputs
as container environment variables and upload the output directory to OSS after
the process exits.
as container environment variables and let the receiver persist the output
after the process exits.
ESP-IDF uses Ninja, which automatically builds in parallel using the CPUs
visible to the container. Allocate at least 8 vCPUs to an ECI build job when
+63 -96
View File
@@ -17,6 +17,9 @@ import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Sequence
@@ -30,8 +33,9 @@ ARTIFACTS = {
"ota": Path("build/xiaozhi.bin"),
"full": Path("build/merged-binary.bin"),
}
OSS_UPLOAD_MAX_ATTEMPTS = 4
OSS_UPLOAD_BASE_DELAY_SECONDS = 1
UPLOAD_MAX_ATTEMPTS = 4
UPLOAD_BASE_DELAY_SECONDS = 1
UPLOAD_TIMEOUT_SECONDS = 120
def utc_now() -> str:
@@ -43,42 +47,19 @@ def env(name: str) -> str | None:
return value.strip() if value and value.strip() else None
def env_enabled(name: str) -> bool:
return (env(name) or "").casefold() in {"1", "true", "yes", "on"}
def oss_config() -> dict[str, str] | None:
if not env_enabled("FIRMWARE_OSS_UPLOAD"):
def upload_config() -> dict[str, str] | None:
upload_url = env("FIRMWARE_UPLOAD_URL")
upload_token = env("FIRMWARE_UPLOAD_TOKEN")
if not upload_url and not upload_token:
return None
values = {
"access_key_id": env("OSS_ACCESS_KEY_ID"),
"access_key_secret": env("OSS_ACCESS_KEY_SECRET"),
"bucket": env("OSS_BUCKET_NAME"),
"endpoint": env("FIRMWARE_OSS_ENDPOINT"),
"prefix": env("FIRMWARE_OSS_PREFIX") or "custom_firmwares",
}
missing = [name for name, value in values.items() if not value]
if missing:
if not upload_url or not upload_token:
raise ValueError(
"OSS upload is enabled but configuration is missing: "
+ ", ".join(missing)
"FIRMWARE_UPLOAD_URL and FIRMWARE_UPLOAD_TOKEN must be configured together"
)
prefix = str(values["prefix"]).strip("/")
if not prefix or ".." in Path(prefix).parts:
raise ValueError(f"Invalid OSS prefix: {prefix!r}")
endpoint = str(values["endpoint"])
if not endpoint.startswith(("http://", "https://")):
endpoint = "https://" + endpoint
return {
"access_key_id": str(values["access_key_id"]),
"access_key_secret": str(values["access_key_secret"]),
"bucket": str(values["bucket"]),
"endpoint": endpoint,
"prefix": prefix,
}
parsed_url = urllib.parse.urlparse(upload_url)
if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
raise ValueError("FIRMWARE_UPLOAD_URL must be an absolute HTTP(S) URL")
return {"url": upload_url.rstrip("/"), "token": upload_token}
def parser() -> argparse.ArgumentParser:
@@ -309,51 +290,54 @@ def failure_summary(log_path: Path) -> str:
return meaningful[-1][:500] if meaningful else "Firmware build failed"
def is_retryable_oss_error(error: Exception, oss2: Any) -> bool:
exceptions = getattr(oss2, "exceptions", None)
request_error = getattr(exceptions, "RequestError", None)
if isinstance(request_error, type) and isinstance(error, request_error):
return True
server_error = getattr(exceptions, "ServerError", None)
if isinstance(server_error, type) and isinstance(error, server_error):
status = getattr(error, "status", getattr(error, "status_code", None))
code = str(getattr(error, "code", ""))
return (
status in {408, 429}
or isinstance(status, int) and status >= 500
or code in {
"InternalError",
"RequestTimeout",
"ServiceUnavailable",
"Throttling",
}
)
return isinstance(error, (ConnectionError, TimeoutError, OSError))
def is_retryable_upload_error(error: Exception) -> bool:
if isinstance(error, urllib.error.HTTPError):
return error.code in {408, 429} or error.code >= 500
return isinstance(
error,
(urllib.error.URLError, ConnectionError, TimeoutError, OSError),
)
def upload_file_with_retry(
bucket: Any,
object_key: str,
upload_url: str,
upload_token: str,
local_path: Path,
oss2: Any,
) -> None:
for attempt in range(1, OSS_UPLOAD_MAX_ATTEMPTS + 1):
payload = local_path.read_bytes()
request = urllib.request.Request(
upload_url,
data=payload,
method="PUT",
headers={
"Authorization": f"Bearer {upload_token}",
"Content-Type": "application/octet-stream",
"Content-Length": str(len(payload)),
"X-Artifact-SHA256": hashlib.sha256(payload).hexdigest(),
},
)
for attempt in range(1, UPLOAD_MAX_ATTEMPTS + 1):
try:
bucket.put_object_from_file(object_key, str(local_path))
with urllib.request.urlopen(
request,
timeout=UPLOAD_TIMEOUT_SECONDS,
) as response:
response.read()
return
except Exception as error:
retryable = is_retryable_upload_error(error)
if isinstance(error, urllib.error.HTTPError):
error.close()
if (
attempt >= OSS_UPLOAD_MAX_ATTEMPTS
or not is_retryable_oss_error(error, oss2)
attempt >= UPLOAD_MAX_ATTEMPTS
or not retryable
):
raise
delay = OSS_UPLOAD_BASE_DELAY_SECONDS * (2 ** (attempt - 1))
delay = UPLOAD_BASE_DELAY_SECONDS * (2 ** (attempt - 1))
print(
"firmware-builder: transient OSS upload failure for "
"firmware-builder: transient artifact upload failure for "
f"{local_path.name}; retry {attempt + 1}/"
f"{OSS_UPLOAD_MAX_ATTEMPTS} in {delay}s: {error}",
f"{UPLOAD_MAX_ATTEMPTS} in {delay}s: {error}",
file=sys.stderr,
)
time.sleep(delay)
@@ -365,49 +349,32 @@ def upload_outputs(
config: dict[str, str],
) -> None:
if not manifest.get("job_id"):
raise ValueError("FIRMWARE_JOB_ID is required when OSS upload is enabled")
try:
import oss2
except ImportError as error:
raise RuntimeError("oss2 is required for OSS upload") from error
raise ValueError("FIRMWARE_JOB_ID is required when artifact upload is enabled")
job_id = str(manifest["job_id"])
if not SAFE_JOB_ID.fullmatch(job_id):
raise ValueError(f"Invalid job ID for OSS upload: {job_id!r}")
raise ValueError(f"Invalid job ID for artifact upload: {job_id!r}")
base_key = f"{config['prefix']}/{job_id}"
file_names = ["build.log"]
file_names.extend(str(item["file"]) for item in manifest["artifacts"])
file_names.append("manifest.json")
object_keys = {name: f"{base_key}/{name}" for name in file_names}
manifest["oss"] = {
"bucket": config["bucket"],
"endpoint": config["endpoint"],
"prefix": base_key,
"objects": object_keys,
}
manifest["delivery_status"] = "uploading"
write_manifest(output_dir, manifest)
auth = oss2.Auth(config["access_key_id"], config["access_key_secret"])
bucket = oss2.Bucket(auth, config["endpoint"], config["bucket"])
for name in file_names[:-1]:
upload_file_with_retry(
bucket,
object_keys[name],
f"{config['url']}/{urllib.parse.quote(job_id)}/artifacts/"
f"{urllib.parse.quote(name)}",
config["token"],
output_dir / name,
oss2,
)
manifest["delivery_status"] = "succeeded"
write_manifest(output_dir, manifest)
upload_file_with_retry(
bucket,
object_keys["manifest.json"],
f"{config['url']}/{urllib.parse.quote(job_id)}/artifacts/manifest.json",
config["token"],
output_dir / "manifest.json",
oss2,
)
@@ -416,7 +383,7 @@ def main(argv: Sequence[str] | None = None) -> int:
started_at = utc_now()
try:
validate(args)
upload_config = oss_config()
artifact_upload_config = upload_config()
except ValueError as error:
print(f"firmware-builder: {error}", file=sys.stderr)
return 2
@@ -475,15 +442,15 @@ def main(argv: Sequence[str] | None = None) -> int:
manifest["error"] = failure_summary(log_path)
write_manifest(args.output_dir, manifest)
if upload_config is not None:
if artifact_upload_config is not None:
try:
print("XIAOZHI_STAGE uploading", flush=True)
upload_outputs(args.output_dir, manifest, upload_config)
upload_outputs(args.output_dir, manifest, artifact_upload_config)
except Exception as error:
print(f"firmware-builder: OSS upload failed: {error}", file=sys.stderr)
print(f"firmware-builder: artifact upload failed: {error}", file=sys.stderr)
manifest["status"] = "failed"
manifest["delivery_status"] = "failed"
manifest["error"] = f"OSS upload failed: {error}"
manifest["error"] = f"Artifact upload failed: {error}"
return_code = 1
manifest["exit_code"] = return_code
write_manifest(args.output_dir, manifest)
@@ -3,7 +3,6 @@ import json
import os
import sys
import tempfile
import types
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -154,31 +153,26 @@ sys.exit(%d)
)
def test_success_uploads_outputs_and_manifest_last(self) -> None:
uploads: list[tuple[str, str]] = []
uploads: list[object] = []
class FakeBucket:
def __init__(self, auth: object, endpoint: str, bucket: str) -> None:
self.auth = auth
self.endpoint = endpoint
self.bucket = bucket
class FakeResponse:
def __enter__(self) -> "FakeResponse":
return self
def put_object_from_file(self, key: str, path: str) -> None:
uploads.append((key, path))
def __exit__(self, *args: object) -> None:
return None
def read(self) -> bytes:
return b""
def fake_urlopen(request: object, timeout: int) -> FakeResponse:
uploads.append(request)
self.assertEqual(timeout, firmware_builder.UPLOAD_TIMEOUT_SECONDS)
return FakeResponse()
fake_oss2 = types.SimpleNamespace(
Auth=lambda access_key_id, access_key_secret: (
access_key_id,
access_key_secret,
),
Bucket=FakeBucket,
)
upload_env = {
"FIRMWARE_OSS_UPLOAD": "true",
"FIRMWARE_OSS_ENDPOINT": "oss-cn-shenzhen.aliyuncs.com",
"FIRMWARE_OSS_PREFIX": "custom_firmwares",
"OSS_BUCKET_NAME": "test-bucket",
"OSS_ACCESS_KEY_ID": "test-id",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"FIRMWARE_UPLOAD_URL": "https://example.com/api/firmware-builds",
"FIRMWARE_UPLOAD_TOKEN": "test-token",
}
with tempfile.TemporaryDirectory() as temporary:
@@ -187,7 +181,7 @@ sys.exit(%d)
output = root / "output"
with (
patch.dict(os.environ, upload_env),
patch.dict(sys.modules, {"oss2": fake_oss2}),
patch.object(firmware_builder.urllib.request, "urlopen", fake_urlopen),
):
exit_code = firmware_builder.main(
[
@@ -210,122 +204,97 @@ sys.exit(%d)
self.assertEqual(exit_code, 0)
self.assertEqual(len(uploads), 4)
self.assertEqual(uploads[-1][0], "custom_firmwares/test-job/manifest.json")
self.assertEqual(
uploads[-1].full_url,
"https://example.com/api/firmware-builds/test-job/artifacts/manifest.json",
)
self.assertEqual(uploads[-1].get_header("Authorization"), "Bearer test-token")
manifest = json.loads((output / "manifest.json").read_text())
self.assertEqual(manifest["delivery_status"], "succeeded")
self.assertNotIn("test-secret", json.dumps(manifest))
self.assertNotIn("test-token", json.dumps(manifest))
def test_transient_oss_upload_is_retried_with_exponential_backoff(self) -> None:
class FakeRequestError(Exception):
pass
def test_transient_upload_is_retried_with_exponential_backoff(self) -> None:
attempts = 0
class FakeBucket:
attempts = 0
def put_object_from_file(self, key: str, path: str) -> None:
self.attempts += 1
if self.attempts < 3:
raise FakeRequestError("connection timed out")
fake_oss2 = types.SimpleNamespace(
exceptions=types.SimpleNamespace(
RequestError=FakeRequestError,
ServerError=type("FakeServerError", (Exception,), {}),
)
)
def fake_urlopen(request: object, timeout: int) -> object:
nonlocal attempts
attempts += 1
if attempts < 3:
raise firmware_builder.urllib.error.URLError("connection timed out")
return unittest.mock.MagicMock()
with tempfile.TemporaryDirectory() as temporary:
local_path = Path(temporary) / "build.log"
local_path.write_text("build output", encoding="utf-8")
bucket = FakeBucket()
with patch.object(firmware_builder.time, "sleep") as sleep:
with (
patch.object(firmware_builder.time, "sleep") as sleep,
patch.object(firmware_builder.urllib.request, "urlopen", fake_urlopen),
):
firmware_builder.upload_file_with_retry(
bucket,
"firmware/test/build.log",
"https://example.com/build.log",
"test-token",
local_path,
fake_oss2,
)
self.assertEqual(bucket.attempts, 3)
self.assertEqual(attempts, 3)
self.assertEqual(
[call.args[0] for call in sleep.call_args_list],
[1, 2],
)
def test_transient_oss_upload_fails_after_retry_limit(self) -> None:
class FakeRequestError(Exception):
pass
def test_transient_upload_fails_after_retry_limit(self) -> None:
attempts = 0
class FakeBucket:
attempts = 0
def put_object_from_file(self, key: str, path: str) -> None:
self.attempts += 1
raise FakeRequestError("connection timed out")
fake_oss2 = types.SimpleNamespace(
exceptions=types.SimpleNamespace(
RequestError=FakeRequestError,
ServerError=type("FakeServerError", (Exception,), {}),
)
)
def fake_urlopen(request: object, timeout: int) -> object:
nonlocal attempts
attempts += 1
raise firmware_builder.urllib.error.URLError("connection timed out")
with tempfile.TemporaryDirectory() as temporary:
local_path = Path(temporary) / "build.log"
local_path.write_text("build output", encoding="utf-8")
bucket = FakeBucket()
with (
patch.object(firmware_builder.time, "sleep") as sleep,
self.assertRaisesRegex(FakeRequestError, "connection timed out"),
patch.object(firmware_builder.urllib.request, "urlopen", fake_urlopen),
self.assertRaisesRegex(firmware_builder.urllib.error.URLError, "connection timed out"),
):
firmware_builder.upload_file_with_retry(
bucket,
"firmware/test/build.log",
"https://example.com/build.log",
"test-token",
local_path,
fake_oss2,
)
self.assertEqual(bucket.attempts, firmware_builder.OSS_UPLOAD_MAX_ATTEMPTS)
self.assertEqual(attempts, firmware_builder.UPLOAD_MAX_ATTEMPTS)
self.assertEqual(
[call.args[0] for call in sleep.call_args_list],
[1, 2, 4],
)
def test_permanent_oss_upload_error_is_not_retried(self) -> None:
class FakeServerError(Exception):
status = 403
code = "AccessDenied"
def test_permanent_upload_error_is_not_retried(self) -> None:
attempts = 0
class FakeBucket:
attempts = 0
def put_object_from_file(self, key: str, path: str) -> None:
self.attempts += 1
raise FakeServerError("access denied")
fake_oss2 = types.SimpleNamespace(
exceptions=types.SimpleNamespace(
RequestError=type("FakeRequestError", (Exception,), {}),
ServerError=FakeServerError,
def fake_urlopen(request: object, timeout: int) -> object:
nonlocal attempts
attempts += 1
raise firmware_builder.urllib.error.HTTPError(
request.full_url, 403, "Forbidden", {}, None
)
)
with tempfile.TemporaryDirectory() as temporary:
local_path = Path(temporary) / "build.log"
local_path.write_text("build output", encoding="utf-8")
bucket = FakeBucket()
with (
patch.object(firmware_builder.time, "sleep") as sleep,
self.assertRaisesRegex(FakeServerError, "access denied"),
patch.object(firmware_builder.urllib.request, "urlopen", fake_urlopen),
self.assertRaisesRegex(firmware_builder.urllib.error.HTTPError, "403"),
):
firmware_builder.upload_file_with_retry(
bucket,
"firmware/test/build.log",
"https://example.com/build.log",
"test-token",
local_path,
fake_oss2,
)
self.assertEqual(bucket.attempts, 1)
self.assertEqual(attempts, 1)
sleep.assert_not_called()
def test_rejects_path_traversal_board(self) -> None:
+1 -1
View File
@@ -722,7 +722,7 @@ endchoice
choice DISPLAY_LCD_TYPE
depends on BOARD_TYPE_BREAD_COMPACT_WIFI_LCD || BOARD_TYPE_BREAD_COMPACT_ESP32_LCD || BOARD_TYPE_CGC || BOARD_TYPE_BREAD_COMPACT_WIFI_CAM || (BOARD_TYPE_WK_ESP32S3_DEV && WK_ESP32S3_DEV_DISPLAY_LCD)
depends on BOARD_TYPE_BREAD_COMPACT_WIFI_LCD || BOARD_TYPE_BREAD_COMPACT_ESP32_LCD || BOARD_TYPE_WDMOMO_CGC || BOARD_TYPE_BREAD_COMPACT_WIFI_CAM || (BOARD_TYPE_WK_ESP32S3_DEV && WK_ESP32S3_DEV_DISPLAY_LCD)
prompt "LCD Type"
default LCD_ST7789_240X320
help
@@ -1,18 +1,22 @@
#include "board.h"
#include "nt26_board.h"
#include "codecs/no_audio_codec.h"
#include "display/oled_display.h"
#include "application.h"
#include "assets/lang_config.h"
#include "board.h"
#include "button.h"
#include "codecs/no_audio_codec.h"
#include "config.h"
#include "display/oled_display.h"
#include "lamp_controller.h"
#include "led/single_led.h"
#include "assets/lang_config.h"
#include "nt26_board.h"
#include <esp_log.h>
#include <driver/i2c_master.h>
#include <esp_lcd_panel_ops.h>
#include <esp_lcd_panel_vendor.h>
#include <esp_log.h>
#ifdef SH1106
#include <esp_lcd_panel_sh1106.h>
#endif
#define TAG "CompactNt26Board"
@@ -36,15 +40,16 @@ private:
.glitch_ignore_cnt = 7,
.intr_priority = 0,
.trans_queue_depth = 0,
.flags = {
.enable_internal_pullup = 1,
},
.flags =
{
.enable_internal_pullup = 1,
},
};
ESP_ERROR_CHECK(i2c_new_master_bus(&bus_config, &display_i2c_bus_));
}
void InitializeSsd1306Display() {
// SSD1306 config
void InitializeOledDisplay() {
// OLED config
esp_lcd_panel_io_i2c_config_t io_config = {
.dev_addr = 0x3C,
.scl_speed_hz = 400 * 1000,
@@ -54,15 +59,16 @@ private:
.lcd_param_bits = 8,
.on_color_trans_done = nullptr,
.user_ctx = nullptr,
.flags = {
.dc_low_on_data = 0,
.disable_control_phase = 0,
},
.flags =
{
.dc_low_on_data = 0,
.disable_control_phase = 0,
},
};
ESP_ERROR_CHECK(esp_lcd_new_panel_io_i2c(display_i2c_bus_, &io_config, &panel_io_));
ESP_LOGI(TAG, "Install SSD1306 driver");
ESP_LOGI(TAG, "Install OLED driver");
esp_lcd_panel_dev_config_t panel_config = {};
panel_config.reset_gpio_num = GPIO_NUM_NC;
panel_config.bits_per_pixel = 1;
@@ -72,8 +78,12 @@ private:
};
panel_config.vendor_config = &ssd1306_config;
#ifdef SH1106
ESP_ERROR_CHECK(esp_lcd_new_panel_sh1106(panel_io_, &panel_config, &panel_));
#else
ESP_ERROR_CHECK(esp_lcd_new_panel_ssd1306(panel_io_, &panel_config, &panel_));
ESP_LOGI(TAG, "SSD1306 driver installed");
#endif
ESP_LOGI(TAG, "OLED driver installed");
// Reset the display
ESP_ERROR_CHECK(esp_lcd_panel_reset(panel_));
@@ -82,25 +92,21 @@ private:
display_ = new NoDisplay();
return;
}
ESP_ERROR_CHECK(esp_lcd_panel_invert_color(panel_, false));
// Set the display to on
ESP_LOGI(TAG, "Turning display on");
ESP_ERROR_CHECK(esp_lcd_panel_disp_on_off(panel_, true));
display_ = new OledDisplay(panel_io_, panel_, DISPLAY_WIDTH, DISPLAY_HEIGHT, DISPLAY_MIRROR_X, DISPLAY_MIRROR_Y);
display_ = new OledDisplay(panel_io_, panel_, DISPLAY_WIDTH, DISPLAY_HEIGHT,
DISPLAY_MIRROR_X, DISPLAY_MIRROR_Y);
}
void InitializeButtons() {
boot_button_.OnClick([]() {
Application::GetInstance().ToggleChatState();
});
boot_button_.OnClick([]() { Application::GetInstance().ToggleChatState(); });
touch_button_.OnPressDown([]() {
Application::GetInstance().StartListening();
});
touch_button_.OnPressUp([]() {
Application::GetInstance().StopListening();
});
touch_button_.OnPressDown([]() { Application::GetInstance().StartListening(); });
touch_button_.OnPressUp([]() { Application::GetInstance().StopListening(); });
volume_up_button_.OnClick([this]() {
auto codec = GetAudioCodec();
@@ -134,20 +140,17 @@ private:
}
// 物联网初始化,添加对 AI 可见设备
void InitializeTools() {
static LampController lamp(LAMP_GPIO);
}
void InitializeTools() { static LampController lamp(LAMP_GPIO); }
public:
CompactNt26Board() :
Nt26Board(NT26_TX_PIN, NT26_RX_PIN, NT26_DTR_PIN, NT26_RI_PIN),
boot_button_(BOOT_BUTTON_GPIO),
touch_button_(TOUCH_BUTTON_GPIO),
volume_up_button_(VOLUME_UP_BUTTON_GPIO),
volume_down_button_(VOLUME_DOWN_BUTTON_GPIO) {
CompactNt26Board()
: Nt26Board(NT26_TX_PIN, NT26_RX_PIN, NT26_DTR_PIN, NT26_RI_PIN),
boot_button_(BOOT_BUTTON_GPIO),
touch_button_(TOUCH_BUTTON_GPIO),
volume_up_button_(VOLUME_UP_BUTTON_GPIO),
volume_down_button_(VOLUME_DOWN_BUTTON_GPIO) {
InitializeDisplayI2c();
InitializeSsd1306Display();
InitializeOledDisplay();
InitializeButtons();
InitializeTools();
}
@@ -165,17 +168,18 @@ public:
virtual AudioCodec* GetAudioCodec() override {
#ifdef AUDIO_I2S_METHOD_SIMPLEX
static NoAudioCodecSimplex audio_codec(AUDIO_INPUT_SAMPLE_RATE, AUDIO_OUTPUT_SAMPLE_RATE,
AUDIO_I2S_SPK_GPIO_BCLK, AUDIO_I2S_SPK_GPIO_LRCK, AUDIO_I2S_SPK_GPIO_DOUT, AUDIO_I2S_MIC_GPIO_SCK, AUDIO_I2S_MIC_GPIO_WS, AUDIO_I2S_MIC_GPIO_DIN);
AUDIO_I2S_SPK_GPIO_BCLK, AUDIO_I2S_SPK_GPIO_LRCK,
AUDIO_I2S_SPK_GPIO_DOUT, AUDIO_I2S_MIC_GPIO_SCK,
AUDIO_I2S_MIC_GPIO_WS, AUDIO_I2S_MIC_GPIO_DIN);
#else
static NoAudioCodecDuplex audio_codec(AUDIO_INPUT_SAMPLE_RATE, AUDIO_OUTPUT_SAMPLE_RATE,
AUDIO_I2S_GPIO_BCLK, AUDIO_I2S_GPIO_WS, AUDIO_I2S_GPIO_DOUT, AUDIO_I2S_GPIO_DIN);
AUDIO_I2S_GPIO_BCLK, AUDIO_I2S_GPIO_WS,
AUDIO_I2S_GPIO_DOUT, AUDIO_I2S_GPIO_DIN);
#endif
return &audio_codec;
}
virtual Display* GetDisplay() override {
return display_;
}
virtual Display* GetDisplay() override { return display_; }
};
DECLARE_BOARD(CompactNt26Board);
+3
View File
@@ -41,6 +41,9 @@
#define DISPLAY_HEIGHT 32
#elif CONFIG_OLED_SSD1306_128X64
#define DISPLAY_HEIGHT 64
#elif CONFIG_OLED_SH1106_128X64
#define DISPLAY_HEIGHT 64
#define SH1106
#else
#error "OLED display type is not selected"
#endif
+14 -1
View File
@@ -77,7 +77,7 @@ esp_err_t esp_lcd_new_panel_ili9486(const esp_lcd_panel_io_handle_t io, const es
ESP_GOTO_ON_FALSE(false, ESP_ERR_NOT_SUPPORTED, err, TAG, "unsupported color space");
break;
}
#else
#elif ESP_IDF_VERSION < ESP_IDF_VERSION_VAL(6, 0, 0)
switch (panel_dev_config->rgb_endian) {
case LCD_RGB_ENDIAN_RGB:
ili9486->madctl_val = 0;
@@ -89,6 +89,19 @@ esp_err_t esp_lcd_new_panel_ili9486(const esp_lcd_panel_io_handle_t io, const es
ESP_GOTO_ON_FALSE(false, ESP_ERR_NOT_SUPPORTED, err, TAG, "unsupported rgb endian");
break;
}
#else
switch (panel_dev_config->rgb_ele_order) {
case LCD_RGB_ELEMENT_ORDER_RGB:
ili9486->madctl_val = 0;
break;
case LCD_RGB_ELEMENT_ORDER_BGR:
ili9486->madctl_val |= LCD_CMD_BGR_BIT;
break;
default:
ESP_GOTO_ON_FALSE(false, ESP_ERR_NOT_SUPPORTED, err, TAG,
"unsupported rgb element order");
break;
}
#endif
switch (panel_dev_config->bits_per_pixel) {
+14 -1
View File
@@ -73,7 +73,7 @@ esp_err_t esp_lcd_new_panel_nv3030b(const esp_lcd_panel_io_handle_t io, const es
ESP_GOTO_ON_FALSE(false, ESP_ERR_NOT_SUPPORTED, err, TAG, "unsupported color space");
break;
}
#else
#elif ESP_IDF_VERSION < ESP_IDF_VERSION_VAL(6, 0, 0)
switch (panel_dev_config->rgb_endian) {
case LCD_RGB_ENDIAN_RGB:
nv3030b->madctl_val = 0;
@@ -85,6 +85,19 @@ esp_err_t esp_lcd_new_panel_nv3030b(const esp_lcd_panel_io_handle_t io, const es
ESP_GOTO_ON_FALSE(false, ESP_ERR_NOT_SUPPORTED, err, TAG, "unsupported rgb endian");
break;
}
#else
switch (panel_dev_config->rgb_ele_order) {
case LCD_RGB_ELEMENT_ORDER_RGB:
nv3030b->madctl_val = 0;
break;
case LCD_RGB_ELEMENT_ORDER_BGR:
nv3030b->madctl_val |= LCD_CMD_BGR_BIT;
break;
default:
ESP_GOTO_ON_FALSE(false, ESP_ERR_NOT_SUPPORTED, err, TAG,
"unsupported rgb element order");
break;
}
#endif
switch (panel_dev_config->bits_per_pixel) {
+15 -16
View File
@@ -101,26 +101,25 @@ private:
void InitializeSsd1306Display() {
// SSD1306 config
esp_lcd_panel_io_i2c_config_t io_config = {
.dev_addr = 0x3C,
.on_color_trans_done = nullptr,
.user_ctx = nullptr,
.control_phase_bytes = 1,
.dc_bit_offset = 6,
.lcd_cmd_bits = 8,
.lcd_param_bits = 8,
.flags = {
.dc_low_on_data = 0,
.disable_control_phase = 0,
},
.scl_speed_hz = 400 * 1000,
};
// IDF 5.5 and 6.x declare these fields in a different order. Assign
// them individually so C++ designated-initializer ordering is irrelevant.
esp_lcd_panel_io_i2c_config_t io_config = {};
io_config.dev_addr = 0x3C;
io_config.scl_speed_hz = 400 * 1000;
io_config.control_phase_bytes = 1;
io_config.dc_bit_offset = 6;
io_config.lcd_cmd_bits = 8;
io_config.lcd_param_bits = 8;
io_config.on_color_trans_done = nullptr;
io_config.user_ctx = nullptr;
io_config.flags.dc_low_on_data = 0;
io_config.flags.disable_control_phase = 0;
ESP_ERROR_CHECK(esp_lcd_new_panel_io_i2c_v2(display_i2c_bus_, &io_config, &panel_io_));
ESP_ERROR_CHECK(esp_lcd_new_panel_io_i2c(display_i2c_bus_, &io_config, &panel_io_));
ESP_LOGI(TAG, "Install SSD1306 driver");
esp_lcd_panel_dev_config_t panel_config = {};
panel_config.reset_gpio_num = -1;
panel_config.reset_gpio_num = GPIO_NUM_NC;
panel_config.bits_per_pixel = 1;
esp_lcd_panel_ssd1306_config_t ssd1306_config = {
+9
View File
@@ -1062,6 +1062,15 @@ class BuildOptionTests(unittest.TestCase):
).read_text(encoding="utf-8")
self.assertIn("CONFIG_OLED_SH1106_128X64", config_header)
def test_bread_compact_nt26_supports_sh1106(self):
board_dir = ROOT / "main/boards/bread-compact-nt26"
config_header = (board_dir / "config.h").read_text(encoding="utf-8")
board_source = (board_dir / "compact_nt26_board.cc").read_text(
encoding="utf-8"
)
self.assertIn("CONFIG_OLED_SH1106_128X64", config_header)
self.assertIn("esp_lcd_new_panel_sh1106", board_source)
def test_non_default_style_disables_multiline_chat(self):
definitions = [
{