ai: вариант 5 генератора на audiopwmio (CircuitPython), задание и брошюры синхронизированы с РПД
This commit is contained in:
mss
@@ -0,0 +1,854 @@
|
||||
# DDS-генератор на аудио-ШИМ (RP2040) — CircuitPython
|
||||
#
|
||||
# Вариант 5 генератора для лабораторных работ. Совместим по системе команд
|
||||
# с вариантами 1–4 (F, SF, W, D, M, A, SWEEP, STOP, B, O, N, FN, FP, I,
|
||||
# CAL, HELP), но выход сделан не через PIO и не через программный ШИМ,
|
||||
# а через штатный звуковой модуль CircuitPython audiopwmio.
|
||||
#
|
||||
# ЧЕРНОВИК (AI-сгенерировано, 2026-10-28). Требует проверки преподавателем
|
||||
# на реальном стенде: частоты и разрядность посчитаны по исходникам ядра,
|
||||
# но реальный размах и уровень несущей надо проверить осциллографом.
|
||||
#
|
||||
# Дополнительные библиотеки НЕ нужны: audiopwmio и audiocore входят в
|
||||
# прошивку CircuitPython для RP2040.
|
||||
#
|
||||
# ──────────────── ПОЧЕМУ НЕ analogio (важно) ────────────────
|
||||
#
|
||||
# В CircuitPython модуль analogio содержит ровно два класса:
|
||||
#
|
||||
# analogio.AnalogIn — вход АЦП (им и меряем, см. команду A);
|
||||
# analogio.AnalogOut — выход ЦАП.
|
||||
#
|
||||
# AnalogOut работает только там, где есть НАСТОЯЩИЙ ЦАП. У RP2040 его нет,
|
||||
# поэтому в документации прямо написано: «Not available on Nordic, RP2040,
|
||||
# Spresense, as there is no on-chip DAC». То есть аналогового выхода на
|
||||
# Pico через analogio не получить — ни в каком виде.
|
||||
#
|
||||
# Вывод «звука через ШИМ» в CircuitPython делает другой модуль — audiopwmio:
|
||||
#
|
||||
# audiopwmio.PWMAudioOut(pin) — канал вывода (ШИМ-слайс);
|
||||
# audiocore.RawSample(буфер, ...) — проигрываемый сэмпл.
|
||||
#
|
||||
# Именно его и использует этот вариант.
|
||||
#
|
||||
# ──────────────── КАК ЭТО РАБОТАЕТ ────────────────
|
||||
#
|
||||
# На RP2040 внутри audiopwmio:
|
||||
#
|
||||
# * занимается один ШИМ-слайс (GP15 — это слайс 7, канал B);
|
||||
# * TOP регистра скважности = 1023, поэтому несущая ШИМ:
|
||||
#
|
||||
# f_несущей = 125 МГц / 1024 = 122,07 кГц (фиксирована ядром)
|
||||
#
|
||||
# * скважность (10 бит из 16-битного сэмпла) обновляется НЕ программой,
|
||||
# а DMA по таймеру темпа. Таймер настраивается дробным делителем
|
||||
# 125 МГц * X / Y, поэтому частота сэмплирования задаётся очень точно:
|
||||
#
|
||||
# sample_rate = f_вых * m, где m = N * O — сэмплов на период
|
||||
#
|
||||
# * после RC-фильтра низких частот остаётся среднее значение скважности,
|
||||
# то есть наш сигнал. Несущие 122 кГц фильтр давит, сигнал пропускает.
|
||||
#
|
||||
# Отличие от варианта 1 и 2 (программный ШИМ) — в том, что цикл Python в
|
||||
# формировании сигнала НЕ участвует вообще. Отличие от вариантов 3 и 4
|
||||
# (PIO) — в том, что вместо PIO-машины работает штатный аудио-тракт ядра.
|
||||
#
|
||||
# ──────────────── ЧТО ЭТО ДАЁТ И ЧЕГО НЕ ДАЁТ ────────────────
|
||||
#
|
||||
# Плюсы:
|
||||
# * вывод идёт в фоне (DMA), CPU свободен — команды принимаются на ходу,
|
||||
# АЦП можно читать ОДНОВРЕМЕННО с выводом сигнала;
|
||||
# * несущая 122 кГц фильтруется тем же RC, что и в остальных вариантах
|
||||
# (1к/47н, f_c ~ 3,4 кГц);
|
||||
# * не нужны adafruit_pioasm и rp2pio — только прошивка.
|
||||
#
|
||||
# Минусы (важно для лаборатории):
|
||||
# * это ЗВУКОВОЙ тракт: частота сэмплирования ограничена сверху ~48 кГц,
|
||||
# потому что она должна быть заметно ниже несущей 122 кГц. Значит
|
||||
# верхняя частота сигнала:
|
||||
#
|
||||
# f_вых_max = sample_rate_max / m
|
||||
#
|
||||
# при m = 16 это 3 кГц, при m = 48 — 1 кГц, при m = 256 — 188 Гц.
|
||||
# Выше 3–4 кГц этот вариант не годится, там нужны PIO (варианты 3–4)
|
||||
# или программный ШИМ (варианты 1–2).
|
||||
# * разрядность скважности 10 бит — команда B больше 10 не имеет смысла
|
||||
# (ядро всё равно оставит 10 старших бит).
|
||||
# * при малом m форма ступенчатая: 16 точек на период — это грубо.
|
||||
#
|
||||
# ──────────────── ПИНЫ ────────────────
|
||||
#
|
||||
# GP15 — выход (ШИМ в режиме Q, аудио-ШИМ в режиме S);
|
||||
# GP16 — синхроимпульс (вне зоны, которую занимает аудио);
|
||||
# GP26 — вход ADC0 (A0), до 3,3 В, больше — только через делитель.
|
||||
#
|
||||
# ──────────────── КОМАНДЫ ────────────────
|
||||
#
|
||||
# F <Гц> — частота (1..200000, но см. предел выше)
|
||||
# SF <Гц> — то же, явно для синуса
|
||||
# W Q | W S — режим: Q — прямоугольник (pwmio), S — звук (audiopwmio)
|
||||
# D <0..255> — скважность прямоугольника в режиме Q
|
||||
# M <0..127> — амплитуда (127 = полная шкала)
|
||||
# A — измерить размах на ADC0 -> "A <мВ>"
|
||||
# SWEEP f0 f1 n ms — свип: "P <Гц> <мВ>" на точку, затем "DONE"
|
||||
# STOP — прервать свип
|
||||
# B <1..10> — разрядность скважности (10 — предел audiopwmio)
|
||||
# O <1..64> — множитель сэмплов: m = N*O
|
||||
# N <16..4096> — «полезных» сэмплов на период
|
||||
# FN <имя> — форма: sin, sine2, triangle, saw, square
|
||||
# FP <выражение> — своя функция от t (0..1)
|
||||
# CAL <Гц> — автокалибровка по осциллографу (поправка частоты)
|
||||
# CLC <n> — НЕ ПРИМЕНИМА (нет PIO); принимается для совместимости
|
||||
# I — настройки, пределы частоты, фактическая частота
|
||||
#
|
||||
# ВНИМАНИЕ: ADC0 (GP26) — вход 0..3,3 В. Сигнал больше 3,3 В подавать
|
||||
# только через делитель. Схему без питания к ADC не подключать.
|
||||
|
||||
import array
|
||||
import math
|
||||
import sys
|
||||
import time
|
||||
|
||||
import board
|
||||
import digitalio
|
||||
import pwmio
|
||||
|
||||
import audiocore
|
||||
import audiopwmio
|
||||
|
||||
try:
|
||||
import analogbufio
|
||||
except ImportError: # pragma: no cover
|
||||
analogbufio = None
|
||||
print("# analogbufio недоступен: команды A и SWEEP работают по-простому")
|
||||
|
||||
|
||||
# ─────────────────────────── константы ───────────────────────────
|
||||
|
||||
PIN_OUT = board.GP15 # выход: PWM в режиме Q, аудио-ШИМ в режиме S
|
||||
PIN_TRIG = board.GP16 # синхроимпульс
|
||||
PIN_ADC = board.GP26 # ADC0 — вход измерения
|
||||
|
||||
SYS_CLK = 125_000_000 # тактовая RP2040, Гц
|
||||
|
||||
# Внутри audiopwmio на RP2040: BITS_PER_SAMPLE = 10, TOP = 2^10 - 1 = 1023.
|
||||
# Отсюда фиксированная несущая ШИМ и предельная разрядность 10 бит.
|
||||
PWM_BITS = 10
|
||||
PWM_CARRIER_HZ = SYS_CLK / (1 << PWM_BITS) # 122 070,3 Гц
|
||||
|
||||
# Темп сэмплирования задаёт DMA-таймер с делителем до 65536, значит снизу
|
||||
# частота ограничена ~1907 Гц (та же цифра, что предел clkdiv у PIO).
|
||||
SR_MIN = SYS_CLK / 65536 # ≈ 1907 Гц
|
||||
# Сверху ограничиваем «по-звуковому»: сэмплы должны быть заметно реже
|
||||
# несущей, иначе ШИМ не успевает их отыграть и на выходе будет мусор.
|
||||
SR_MAX = 48000
|
||||
|
||||
ADC_FULL_MV = 3300 # опорное напряжение ADC, мВ
|
||||
|
||||
# Буфер сэмплов: 2 байта на сэмпл ("H"), держим не больше 64 КБ.
|
||||
MAX_BUFFER_BYTES = 64 * 1024
|
||||
|
||||
# Пределы для команд N и O (те же, что в вариантах 3–4, для совместимости).
|
||||
N_MIN, N_MAX = 16, 4096
|
||||
O_MIN, O_MAX = 1, 64
|
||||
|
||||
|
||||
def _default_wave(t):
|
||||
"""Форма по умолчанию: синус. t — доля периода [0,1)."""
|
||||
return math.sin(2.0 * math.pi * t)
|
||||
|
||||
|
||||
# ─────────────────────────── класс AudioPWMDDS ───────────────────────────
|
||||
|
||||
class AudioPWMDDS:
|
||||
"""DDS на audiopwmio: буфер сэмплов играет DMA, CPU свободен."""
|
||||
|
||||
def __init__(self, bits=8, oversample=1, n=32, freq=1000, level=127):
|
||||
self.bits = int(bits)
|
||||
self.oversample = int(oversample)
|
||||
self.n = int(n)
|
||||
self.freq = int(freq)
|
||||
self.level = int(level)
|
||||
self.wave = _default_wave
|
||||
self.wave_name = "sin"
|
||||
self.corr = 1.0 # поправка частоты после CAL
|
||||
self.audio = None # создаётся лениво, в ensure()
|
||||
self._sample = None # RawSample — держим ссылку!
|
||||
self._buf = None # array('H') — буфер должен жить
|
||||
|
||||
# ─── параметры ───
|
||||
|
||||
def samples_per_period(self):
|
||||
"""m — сколько сэмплов приходится на один период сигнала."""
|
||||
return self.n * self.oversample
|
||||
|
||||
def sample_rate_wanted(self):
|
||||
"""Желаемая частота сэмплирования, Гц (до ограничения)."""
|
||||
return self.freq * self.samples_per_period() * self.corr
|
||||
|
||||
def sample_rate_used(self):
|
||||
"""Частота сэмплирования, которая реально будет выставлена."""
|
||||
sr = int(round(self.sample_rate_wanted()))
|
||||
if sr < int(SR_MIN) + 1:
|
||||
sr = int(SR_MIN) + 1
|
||||
if sr > SR_MAX:
|
||||
sr = SR_MAX
|
||||
return sr
|
||||
|
||||
def f_out_actual(self):
|
||||
"""Фактическая частота сигнала при текущем sample_rate."""
|
||||
return self.sample_rate_used() / float(self.samples_per_period())
|
||||
|
||||
def f_out_max(self):
|
||||
return SR_MAX / float(self.samples_per_period())
|
||||
|
||||
def f_out_min(self):
|
||||
return (int(SR_MIN) + 1) / float(self.samples_per_period())
|
||||
|
||||
def needs_warning(self):
|
||||
"""True, если запрошенная частота не влезает в звуковой тракт."""
|
||||
sr = self.sample_rate_wanted()
|
||||
return sr > SR_MAX or sr < SR_MIN
|
||||
|
||||
def suggest_reduction(self):
|
||||
"""Подсказка, что менять, если частота не достижима."""
|
||||
m = self.samples_per_period()
|
||||
if self.sample_rate_wanted() > SR_MAX:
|
||||
ok = int(SR_MAX / float(self.freq * self.corr)) if self.freq else 0
|
||||
return ("уменьшите N*O до %d (сейчас m=%d) или снизьте частоту "
|
||||
"до %.1f Гц" % (max(N_MIN, ok), m, self.f_out_max()))
|
||||
return ("увеличьте N*O (сейчас m=%d): минимум %.1f Гц"
|
||||
% (m, self.f_out_min()))
|
||||
|
||||
def set_frequency(self, hz):
|
||||
self.freq = max(1, min(200000, int(hz)))
|
||||
if self.audio is not None:
|
||||
self._reload()
|
||||
|
||||
def set_bits(self, b):
|
||||
b = int(b)
|
||||
if b < 1:
|
||||
raise ValueError("B должно быть 1..%d" % PWM_BITS)
|
||||
if b > PWM_BITS:
|
||||
raise ValueError("для audiopwmio максимум B=%d (ядро оставляет "
|
||||
"10 старших бит)" % PWM_BITS)
|
||||
self.bits = b
|
||||
if self.audio is not None:
|
||||
self._reload()
|
||||
|
||||
def set_oversample(self, o):
|
||||
o = int(o)
|
||||
if o < O_MIN or o > O_MAX:
|
||||
raise ValueError("O должно быть %d..%d" % (O_MIN, O_MAX))
|
||||
self._check_buffer(self.n * o) # проверить ДО присваивания
|
||||
self.oversample = o
|
||||
if self.audio is not None:
|
||||
self._reload()
|
||||
|
||||
def set_n(self, n):
|
||||
n = int(n)
|
||||
if n < N_MIN or n > N_MAX:
|
||||
raise ValueError("N должно быть %d..%d" % (N_MIN, N_MAX))
|
||||
self._check_buffer(n * self.oversample)
|
||||
self.n = n
|
||||
if self.audio is not None:
|
||||
self._reload()
|
||||
|
||||
def set_amplitude(self, level):
|
||||
self.level = max(0, min(127, int(level)))
|
||||
if self.audio is not None:
|
||||
self._reload()
|
||||
|
||||
def set_function(self, fn, name="func"):
|
||||
self.wave = fn
|
||||
self.wave_name = name
|
||||
if self.audio is not None:
|
||||
self._reload()
|
||||
|
||||
def reset_function(self):
|
||||
self.set_function(_default_wave, "sin")
|
||||
|
||||
def slew_warning(self):
|
||||
"""В этом варианте нет дельта-сигмы, поэтому ограничения наклона нет.
|
||||
Метод оставлен для совместимости с вызовами из main()."""
|
||||
return None
|
||||
|
||||
# ─── буфер и воспроизведение ───
|
||||
|
||||
def _check_buffer(self, m=None):
|
||||
"""Проверить, что буфер для m сэмплов влезает в память."""
|
||||
if m is None:
|
||||
m = self.samples_per_period()
|
||||
if m * 2 > MAX_BUFFER_BYTES:
|
||||
raise ValueError("буфер %d байт > %d — уменьшите N или O"
|
||||
% (m * 2, MAX_BUFFER_BYTES))
|
||||
|
||||
def _build(self):
|
||||
"""Заполнить буфер: m сэмплов одного периода, 16 бит, 0..65535.
|
||||
|
||||
Квантование по B бит делается ЗДЕСЬ, а не в железе: уровень
|
||||
округляется до 2^B значений скважности и растягивается на всю
|
||||
16-битную шкалу. Так команда B честно показывает, сколько
|
||||
ступенек увидит осциллограф.
|
||||
"""
|
||||
m = self.samples_per_period()
|
||||
self._check_buffer(m)
|
||||
|
||||
levels = (1 << self.bits) - 1 # 1, 3, 7, ... 1023
|
||||
amp = self.level / 127.0
|
||||
|
||||
buf = array.array("H", [0]) * m
|
||||
for i in range(m):
|
||||
u = self.wave(i / float(m)) * amp
|
||||
if u > 1.0:
|
||||
u = 1.0
|
||||
elif u < -1.0:
|
||||
u = -1.0
|
||||
# 0.5 + 0.5*u -> 0..1; округляем до B-битной ступеньки
|
||||
level = int(round((0.5 + 0.5 * u) * levels))
|
||||
buf[i] = level * 65535 // levels
|
||||
self._buf = buf
|
||||
return m
|
||||
|
||||
def _reload(self):
|
||||
"""Пересобрать буфер и перезапустить воспроизведение."""
|
||||
self._build()
|
||||
sr = self.sample_rate_used()
|
||||
self._sample = audiocore.RawSample(self._buf,
|
||||
channel_count=1,
|
||||
sample_rate=sr)
|
||||
if self.audio is not None:
|
||||
self.audio.play(self._sample, loop=True)
|
||||
|
||||
def ensure(self):
|
||||
"""Создать канал вывода и запустить сигнал (лениво)."""
|
||||
if self.audio is None:
|
||||
self.audio = audiopwmio.PWMAudioOut(PIN_OUT,
|
||||
quiescent_value=0x8000)
|
||||
self._reload()
|
||||
return self
|
||||
|
||||
def release(self):
|
||||
"""Остановить и освободить ШИМ-слайс (нужно для режима Q)."""
|
||||
if self.audio is not None:
|
||||
try:
|
||||
self.audio.stop()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
self.audio.deinit()
|
||||
except Exception:
|
||||
pass
|
||||
self.audio = None
|
||||
self._sample = None
|
||||
self._buf = None
|
||||
|
||||
def deinit(self):
|
||||
self.release()
|
||||
|
||||
def calibrate(self, measured_hz):
|
||||
"""CAL <измеренная_Гц>: подстроить частоту сэмплирования.
|
||||
|
||||
Ошибка тут может быть только от кварца (доли промилле), но команда
|
||||
оставлена для единообразия с вариантами 3–4: вводим измеренное,
|
||||
получаем поправку corr и пересчёт.
|
||||
"""
|
||||
if self.needs_warning():
|
||||
raise ValueError("сначала задайте достижимую частоту: сейчас "
|
||||
"запрошено %d Гц при пределе %.1f Гц"
|
||||
% (self.freq, self.f_out_max()))
|
||||
measured = float(measured_hz)
|
||||
if measured <= 0:
|
||||
raise ValueError("измеренная частота должна быть > 0")
|
||||
old = self.corr
|
||||
self.corr = self.corr * self.freq / measured
|
||||
self.corr = max(0.5, min(2.0, self.corr))
|
||||
if self.audio is not None:
|
||||
self._reload()
|
||||
return old, self.corr
|
||||
|
||||
def info(self):
|
||||
m = self.samples_per_period()
|
||||
return ("аудио-ШИМ: f=%d Гц, форма=%s, B=%d, N=%d, O=%d, m=%d, "
|
||||
"Fs=%d Гц, несущая=%.1f кГц, f_ном=%.2f Гц, "
|
||||
"поправка=%.6f, f_min=%.1f, f_max=%.1f Гц"
|
||||
% (self.freq, self.wave_name, self.bits, self.n,
|
||||
self.oversample, m, self.sample_rate_used(),
|
||||
PWM_CARRIER_HZ / 1000.0, self.f_out_actual(), self.corr,
|
||||
self.f_out_min(), self.f_out_max()))
|
||||
|
||||
|
||||
# ───────────────────── прямоугольник на PWM ─────────────────────
|
||||
#
|
||||
# Режим Q — обычный аппаратный ШИМ на том же GP15. Одновременно с
|
||||
# аудио-ШИМ он работать не может: audiopwmio занимает весь слайс.
|
||||
# Поэтому режимы включаются по очереди (ensure/release), как в варианте 3.
|
||||
|
||||
class SquarePWM:
|
||||
def __init__(self, pin=PIN_OUT):
|
||||
self.pwm = pwmio.PWMOut(pin, frequency=1000,
|
||||
duty_cycle=0, variable_frequency=True)
|
||||
|
||||
def set_frequency(self, f):
|
||||
self.pwm.frequency = max(1, min(200000, int(f)))
|
||||
|
||||
def set_duty(self, d):
|
||||
d = max(0, min(255, int(d)))
|
||||
self.pwm.duty_cycle = d * 65535 // 255
|
||||
|
||||
def off(self):
|
||||
self.pwm.duty_cycle = 0
|
||||
|
||||
def deinit(self):
|
||||
try:
|
||||
self.pwm.deinit()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ─────────────────────────── АЦП ───────────────────────────
|
||||
|
||||
def measure_amplitude_mv(window_ms=50, rate=100000):
|
||||
"""Размах (пик-пик) на ADC0 в милливольтах.
|
||||
|
||||
Чтение через analogbufio (аппаратный ADC + DMA). Если библиотеки нет
|
||||
или DMA занят звуком — откатываемся на analogio.AnalogIn.
|
||||
|
||||
ВАЖНО: воспроизведение при этом не останавливается — audiopwmio играет
|
||||
на своём DMA-канале, ADC работает независимо.
|
||||
"""
|
||||
if analogbufio is not None:
|
||||
length = int(rate * window_ms / 1000)
|
||||
length = max(64, min(8192, length))
|
||||
buf = array.array("H", [0]) * length
|
||||
adc = None
|
||||
try:
|
||||
adc = analogbufio.BufferedIn(PIN_ADC, sample_rate=rate)
|
||||
adc.readinto(buf, loop=False)
|
||||
return (max(buf) - min(buf)) * ADC_FULL_MV // 65535
|
||||
except Exception:
|
||||
pass # нет DMA-канала или частоты — ниже
|
||||
finally:
|
||||
if adc is not None:
|
||||
try:
|
||||
adc.deinit()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
import analogio
|
||||
adc = analogio.AnalogIn(PIN_ADC)
|
||||
vmin, vmax = 65535, 0
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
while (time.monotonic() - t0) * 1000 < window_ms:
|
||||
v = adc.value
|
||||
if v < vmin:
|
||||
vmin = v
|
||||
if v > vmax:
|
||||
vmax = v
|
||||
finally:
|
||||
adc.deinit()
|
||||
return (vmax - vmin) * ADC_FULL_MV // 65535
|
||||
|
||||
|
||||
# ─────────────────────────── свип ───────────────────────────
|
||||
|
||||
trig = None
|
||||
|
||||
|
||||
def ensure_trig():
|
||||
"""Создать синхроимпульс при первом использовании (лениво!)."""
|
||||
global trig
|
||||
if trig is None:
|
||||
trig = digitalio.DigitalInOut(PIN_TRIG)
|
||||
trig.direction = digitalio.Direction.OUTPUT
|
||||
trig.value = False
|
||||
return trig
|
||||
|
||||
|
||||
def trigger_pulse():
|
||||
t = ensure_trig()
|
||||
t.value = True
|
||||
time.sleep(0.00005)
|
||||
t.value = False
|
||||
|
||||
|
||||
def run_sweep(dds, square, mode, f0, f1, count, dwell_ms):
|
||||
"""Логарифмический свип. Печатает 'P <Гц> <мВ>', в конце 'DONE'."""
|
||||
count = max(2, count)
|
||||
if f1 <= f0:
|
||||
f1 = f0 * 10.0
|
||||
ratio = (f1 / f0) ** (1.0 / (count - 1))
|
||||
f = float(f0)
|
||||
try:
|
||||
for _ in range(count):
|
||||
if mode == "S":
|
||||
dds.set_frequency(int(f))
|
||||
else:
|
||||
square.set_frequency(int(f))
|
||||
trigger_pulse()
|
||||
time.sleep(dwell_ms / 1000.0)
|
||||
mv = measure_amplitude_mv()
|
||||
sys.stdout.write("P %d %d\n" % (int(f), mv))
|
||||
f *= ratio
|
||||
except KeyboardInterrupt:
|
||||
sys.stdout.write("DONE\n")
|
||||
raise
|
||||
sys.stdout.write("DONE\n")
|
||||
|
||||
|
||||
# ───────── предустановленные формы сигнала (для команды FN) ─────────
|
||||
|
||||
def _wave_sin(t):
|
||||
return math.sin(2.0 * math.pi * t)
|
||||
|
||||
|
||||
def _wave_sine2(t):
|
||||
return (0.3 * math.sin(2.0 * math.pi * t) +
|
||||
0.5 * math.sin(4.0 * math.pi * t))
|
||||
|
||||
|
||||
def _wave_triangle(t):
|
||||
return 4.0 * abs(t - 0.5) - 1.0
|
||||
|
||||
|
||||
def _wave_saw(t):
|
||||
return 2.0 * t - 1.0
|
||||
|
||||
|
||||
def _wave_square(t):
|
||||
return 1.0 if t < 0.5 else -1.0
|
||||
|
||||
|
||||
WAVES = {
|
||||
"sin": _wave_sin,
|
||||
"sine2": _wave_sine2,
|
||||
"triangle": _wave_triangle,
|
||||
"saw": _wave_saw,
|
||||
"square": _wave_square,
|
||||
}
|
||||
|
||||
# Разрешённые имена для команд FP/FN: только математика, без встроенных.
|
||||
_EXPR_NAMES = {
|
||||
"sin": math.sin, "cos": math.cos, "tan": math.tan,
|
||||
"pi": math.pi, "abs": abs,
|
||||
}
|
||||
|
||||
EXPR_LEN_MAX = 120
|
||||
|
||||
|
||||
def _eval_expr(expr, t):
|
||||
"""Вычислить выражение от t. Отдельная функция, не lambda: на плате
|
||||
разбор lambda в командах не срабатывал (см. вариант 3)."""
|
||||
env = dict(_EXPR_NAMES)
|
||||
env["t"] = t
|
||||
return eval(expr, {"__builtins__": {}}, env)
|
||||
|
||||
|
||||
def _expr_ok(expr):
|
||||
"""Проверить выражение до использования. None, если всё в порядке."""
|
||||
if len(expr) > EXPR_LEN_MAX:
|
||||
return "выражение слишком длинное (лимит %d)" % EXPR_LEN_MAX
|
||||
for t in (0.0, 0.125, 0.333, 0.5, 0.75):
|
||||
try:
|
||||
v = _eval_expr(expr, t)
|
||||
except Exception as e:
|
||||
return "%s (при t=%.3f)" % (e, t)
|
||||
if v != v: # NaN
|
||||
return "не числовой результат (NaN)"
|
||||
if abs(v) > 1e6:
|
||||
return "слишком большое значение (%g) — форма вне [-1,1]" % v
|
||||
return None
|
||||
|
||||
|
||||
def _make_expr_wave(expr):
|
||||
def wave(t):
|
||||
return _eval_expr(expr, t)
|
||||
return wave
|
||||
|
||||
|
||||
# ─────────────────────────── подробный хелп ───────────────────────────
|
||||
|
||||
def print_help():
|
||||
lines = [
|
||||
"# ─────── DDS-генератор на аудио-ШИМ (audiopwmio): команды ───────",
|
||||
"# Обозначения: <..> — обязательный аргумент.",
|
||||
"#",
|
||||
"# РЕЖИМ И ЧАСТОТА",
|
||||
"# W Q — режим прямоугольника (аппаратный ШИМ на GP15)",
|
||||
"# W S — режим звука (audiopwmio, GP15)",
|
||||
"# F <Гц> — частота, в режиме Q — прямоугольника,",
|
||||
"# в S — сигнала. пример: F 440",
|
||||
"# SF <Гц> — частота СИГНАЛА (то же, явно)",
|
||||
"#",
|
||||
"# ФОРМА СИГНАЛА",
|
||||
"# D <0..255> — скважность прямоугольника (128 = 50 %)",
|
||||
"# M <0..127> — амплитуда (127 = полная шкала)",
|
||||
"# FN <имя> — форма: sin, sine2, triangle, saw, square",
|
||||
"# FP <выражение> — своя функция от t (0..1), например:",
|
||||
"# FP 0.3*sin(2*pi*t)+0.5*sin(4*pi*t)",
|
||||
"#",
|
||||
"# ИЗМЕРЕНИЕ И СВИП (работает ОДНОВРЕМЕННО с выводом)",
|
||||
"# A — размах на ADC0 (GP26), мВ -> 'A <мВ>'",
|
||||
"# SWEEP <f0> <f1> <n> <dwell_ms>",
|
||||
"# логарифмический свип; печатает 'P <Гц> <мВ>',",
|
||||
"# в конце 'DONE'. пример: SWEEP 20 3000 20 150",
|
||||
"# STOP — прервать свип",
|
||||
"#",
|
||||
"# ПАРАМЕТРЫ СИНТЕЗА",
|
||||
"# B <1..10> — разрядность скважности. 10 — предел ядра",
|
||||
"# (внутри audiopwmio берутся 10 старших бит)",
|
||||
"# O <1..64> — множитель сэмплов, m = N * O",
|
||||
"# N <16..4096> — «полезных» сэмплов на период",
|
||||
"# I — настройки, пределы, f_min/f_max, поправка",
|
||||
"# CAL <Гц> — автокалибровка: подайте частоту, измерьте",
|
||||
"# осциллографом, введите измеренное (CAL 440)",
|
||||
"# CLC <n> — НЕ ПРИМЕНИМА: PIO здесь нет; принимается",
|
||||
"# только чтобы не ломать общий сценарий команд",
|
||||
"#",
|
||||
"# ГЛАВНОЕ ОГРАНИЧЕНИЕ (в чём отличие от вариантов 3–4)",
|
||||
"# Частота сэмплирования не больше 48000 Гц, поэтому",
|
||||
"# f_max = 48000 / m.",
|
||||
"# При m=16 это 3 кГц, при m=48 — 1 кГц, при m=256 — 188 Гц.",
|
||||
"# Выше 3–4 кГц этот вариант не работает — берите PIO.",
|
||||
"# Команда I показывает f_min и f_max для текущих N и O.",
|
||||
"#",
|
||||
"# КАРТА ПИНОВ",
|
||||
"# GP15 — выход (ШИМ в Q, аудио-ШИМ в S)",
|
||||
"# GP16 — синхроимпульс",
|
||||
"# GP26 (A0) — вход ADC, до 3,3 В (через делитель)",
|
||||
"#",
|
||||
"# ЧЕМ ЭТО ОТЛИЧАЕТСЯ ОТ analogio",
|
||||
"# analogio умеет только AnalogIn (вход) и AnalogOut (ЦАП).",
|
||||
"# На RP2040 ЦАПА НЕТ, поэтому AnalogOut недоступен, и «звук через",
|
||||
"# ШИМ» делается модулем audiopwmio — он и используется здесь.",
|
||||
"#",
|
||||
"# ФИЛЬТР НА ВЫХОДЕ (тот же, что в остальных вариантах)",
|
||||
"# GP15 --[R1]--+--[R2]--+--> во вход схемы",
|
||||
"# | |",
|
||||
"# [C1] [C2] R1=R2=1к, C1=C2=47н",
|
||||
"# | | (f_c ~ 3.4 кГц)",
|
||||
"# GND GND",
|
||||
"# Несущая ШИМ 122 кГц давится этим фильтром примерно на 60 дБ.",
|
||||
"#",
|
||||
"# СПРАВКА",
|
||||
"# HELP | ? — этот текст",
|
||||
]
|
||||
for ln in lines:
|
||||
sys.stdout.write(ln + "\n")
|
||||
|
||||
|
||||
# ─────────────────────────── main ───────────────────────────
|
||||
|
||||
def main():
|
||||
print("# DDS на аудио-ШИМ (RP2040 / CircuitPython).")
|
||||
print("# Вывод: audiopwmio, вход: analogio/analogbufio.")
|
||||
print("# Подробный хелп: HELP или ?")
|
||||
|
||||
dds = None
|
||||
square = None
|
||||
try:
|
||||
dds = AudioPWMDDS(bits=8, oversample=1, n=32, freq=1000, level=127)
|
||||
square = SquarePWM()
|
||||
square.set_frequency(1000)
|
||||
square.set_duty(128)
|
||||
print("# " + dds.info())
|
||||
print(" (режим Q: PWM на GP15; режим S: audiopwmio на GP15)")
|
||||
except Exception as e:
|
||||
print("! ОШИБКА инициализации:", repr(e))
|
||||
if dds is not None:
|
||||
dds.deinit()
|
||||
if square is not None:
|
||||
square.deinit()
|
||||
return
|
||||
|
||||
mode = "Q"
|
||||
saved_square_hz = 1000
|
||||
|
||||
try:
|
||||
while True:
|
||||
line = sys.stdin.readline()
|
||||
if not line:
|
||||
break # EOF: порт закрыт
|
||||
parts = line.split()
|
||||
if not parts:
|
||||
continue
|
||||
cmd = parts[0].upper()
|
||||
|
||||
if cmd in ("HELP", "?"):
|
||||
print_help()
|
||||
|
||||
elif cmd == "F" and len(parts) > 1:
|
||||
hz = int(float(parts[1]))
|
||||
if mode == "S":
|
||||
dds.set_frequency(hz)
|
||||
print("OK F %d (сигнал)" % dds.freq)
|
||||
if dds.needs_warning():
|
||||
print("! ВНИМАНИЕ: частота недостижима — %s"
|
||||
% dds.suggest_reduction())
|
||||
print("! Сейчас реально: %.1f Гц" % dds.f_out_actual())
|
||||
else:
|
||||
square.set_frequency(hz)
|
||||
print("OK F %d (прямоугольник)" % hz)
|
||||
|
||||
elif cmd == "SF" and len(parts) > 1:
|
||||
dds.set_frequency(int(float(parts[1])))
|
||||
print("OK SF %d" % dds.freq)
|
||||
if dds.needs_warning():
|
||||
print("! ВНИМАНИЕ: частота недостижима — %s"
|
||||
% dds.suggest_reduction())
|
||||
print("! Сейчас реально: %.1f Гц" % dds.f_out_actual())
|
||||
|
||||
elif cmd == "W" and len(parts) > 1:
|
||||
want = "S" if parts[1].upper().startswith("S") else "Q"
|
||||
if want == mode:
|
||||
print("OK W %s (уже)" % mode)
|
||||
else:
|
||||
try:
|
||||
if want == "S":
|
||||
# запомнить частоту прямоугольника, погасить его,
|
||||
# освободить слайс и запустить аудио-ШИМ
|
||||
saved_square_hz = square.pwm.frequency
|
||||
square.off()
|
||||
square.deinit()
|
||||
square = None
|
||||
dds.ensure()
|
||||
mode = "S"
|
||||
print("OK W S (audiopwmio: %s)" % dds.info())
|
||||
else:
|
||||
# остановить аудио, вернуть ШИМ на тот же GP15
|
||||
dds.release()
|
||||
square = SquarePWM()
|
||||
square.set_frequency(saved_square_hz)
|
||||
square.set_duty(128)
|
||||
mode = "Q"
|
||||
print("OK W Q (PWM на GP15, %d Гц)"
|
||||
% saved_square_hz)
|
||||
except Exception as e:
|
||||
print("ERR W %s:" % want, repr(e))
|
||||
|
||||
elif cmd == "D" and len(parts) > 1:
|
||||
if mode == "Q" and square is not None:
|
||||
square.set_duty(int(parts[1]))
|
||||
print("OK D %s" % parts[1])
|
||||
|
||||
elif cmd == "M" and len(parts) > 1:
|
||||
dds.set_amplitude(int(parts[1]))
|
||||
print("OK M %d" % dds.level)
|
||||
|
||||
elif cmd == "A":
|
||||
print("A %d" % measure_amplitude_mv())
|
||||
|
||||
elif cmd == "SWEEP" and len(parts) > 4:
|
||||
run_sweep(dds, square, mode,
|
||||
float(parts[1]), float(parts[2]),
|
||||
int(parts[3]), int(parts[4]))
|
||||
|
||||
elif cmd == "STOP":
|
||||
print("OK STOP (если свип идёт — надёжнее Ctrl-C)")
|
||||
|
||||
elif cmd == "B" and len(parts) > 1:
|
||||
try:
|
||||
dds.set_bits(int(parts[1]))
|
||||
print("OK B %d" % dds.bits)
|
||||
except Exception as e:
|
||||
print("ERR B:", repr(e))
|
||||
|
||||
elif cmd == "O" and len(parts) > 1:
|
||||
try:
|
||||
dds.set_oversample(int(parts[1]))
|
||||
print("OK O %d" % dds.oversample)
|
||||
except Exception as e:
|
||||
print("ERR O:", repr(e))
|
||||
|
||||
elif cmd == "N" and len(parts) > 1:
|
||||
try:
|
||||
dds.set_n(int(parts[1]))
|
||||
print("OK N %d" % dds.n)
|
||||
except Exception as e:
|
||||
print("ERR N:", repr(e))
|
||||
|
||||
elif cmd == "FN" and len(parts) > 1:
|
||||
name = parts[1].lower()
|
||||
if name in WAVES:
|
||||
try:
|
||||
dds.set_function(WAVES[name], name=name)
|
||||
print("OK FN %s" % name)
|
||||
except Exception as e:
|
||||
print("ERR FN:", repr(e))
|
||||
else:
|
||||
print("ERR FN: неизвестная форма. Доступны: %s"
|
||||
% ", ".join(sorted(WAVES)))
|
||||
|
||||
elif cmd == "FP" and len(parts) > 1:
|
||||
expr = line.strip()[len("FP"):].strip()
|
||||
if expr:
|
||||
err = _expr_ok(expr)
|
||||
if err:
|
||||
print("ERR FP: %s" % err)
|
||||
else:
|
||||
try:
|
||||
dds.set_function(_make_expr_wave(expr),
|
||||
name="FP:" + expr)
|
||||
print("OK FP %s" % expr)
|
||||
except Exception as e:
|
||||
print("ERR FP:", repr(e))
|
||||
dds.reset_function()
|
||||
else:
|
||||
print("ERR FP: укажите выражение, например "
|
||||
"FP 0.3*sin(2*pi*t)+0.5*sin(4*pi*t)")
|
||||
|
||||
elif cmd == "CLC" and len(parts) > 1:
|
||||
# Команда есть в вариантах 3–4 (тактов PIO на сэмпл).
|
||||
# Здесь PIO нет: частоту задаёт DMA-таймер, подстраивать
|
||||
# нечего. Принимаем, чтобы общий сценарий не падал.
|
||||
print("OK CLC %s (не применима: здесь нет PIO; для поправки "
|
||||
"частоты — CAL)" % parts[1])
|
||||
|
||||
elif cmd == "CAL" and len(parts) > 1:
|
||||
try:
|
||||
was, now = dds.calibrate(float(parts[1]))
|
||||
print("OK CAL: поправка %.6f -> %.6f (Fs=%d Гц, "
|
||||
"физическая частота должна стать %d Гц)"
|
||||
% (was, now, dds.sample_rate_used(), dds.freq))
|
||||
except Exception as e:
|
||||
print("ERR CAL:", repr(e))
|
||||
|
||||
elif cmd == "I":
|
||||
print("# " + dds.info())
|
||||
|
||||
else:
|
||||
print("ERR unknown command: %s" % cmd)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
dds.deinit()
|
||||
if square is not None:
|
||||
square.deinit()
|
||||
if trig is not None:
|
||||
try:
|
||||
trig.deinit()
|
||||
except Exception:
|
||||
pass
|
||||
print("stopped")
|
||||
|
||||
|
||||
# ─────────────── точка входа ───────────────
|
||||
#
|
||||
# Как и в других вариантах, запускаем безусловно: в CircuitPython проверка
|
||||
# `if __name__ == "__main__":` при автозапуске code.py ненадёжна.
|
||||
|
||||
_START_GENERATOR = True
|
||||
|
||||
if _START_GENERATOR:
|
||||
try:
|
||||
main()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
except Exception as _e:
|
||||
try:
|
||||
import sys as _sys
|
||||
_sys.print_exception(_e)
|
||||
except Exception:
|
||||
print("! ОШИБКА:", repr(_e))
|
||||
Reference in New Issue
Block a user