327 lines
13 KiB
Python
327 lines
13 KiB
Python
# DDS-генератор сигналов для лабораторных работ — версия CircuitPython
|
||
# Платформа: Raspberry Pi Pico / RP2040 с прошивкой CircuitPython
|
||
# (https://circuitpython.org/downloads — выбрать "Raspberry Pi Pico")
|
||
#
|
||
# ЧЕРНОВИК (AI-сгенерировано, 2026-10-26). Требует проверки преподавателем
|
||
# на реальном стенде: номиналы RC-фильтра, диапазоны частот, точность ADC.
|
||
#
|
||
# Чем отличается от Arduino-версии (stub/01-dds-generator-rp2040.ino):
|
||
# - CircuitPython — интерпретатор: «настоящего» таймерного ISR на 62,5 кГц
|
||
# нет, поэтому частота выборок f_s ниже и зависит от скорости Python.
|
||
# Архитектура — без потоков (threading есть не на всех портах):
|
||
# один цикл одновременно обновляет выборку синуса и опрашивает USB-консоль
|
||
# (supervisor.runtime.serial_bytes_available). f_s измеряется на старте
|
||
# (обычно 3–40 кГц на RP2040) и печатается; свип держите до f1 <= f_s/6.
|
||
# - прямоугольник от 20 Гц — аппаратный ШИМ pwmio (точная частота и
|
||
# скважность, как в Arduino); ниже 20 Гц — «программный» прямоугольник
|
||
# в основном цикле;
|
||
# - команды те же: F / W / D / M / A / SWEEP / STOP (через USB-консоль,
|
||
# терминал: 115200).
|
||
#
|
||
# УСТОЙЧИВОСТЬ: ошибки не роняют скетч в REPL — они печатаются в консоль
|
||
# (строки "! ..."), а генератор продолжает работать. Каждая принятая строка
|
||
# логируется как "-> ..." — видно, что реально пришло с терминала.
|
||
#
|
||
# Протокол команд (ввод строкой в терминале, Enter = конец команды):
|
||
# F <Гц> — установить частоту (1..200000)
|
||
# SF <Гц> — частота синусоиды (1..f_s/2, защита от алиасинга)
|
||
# W Q | W S — режим: Q — прямоугольник, S — синус
|
||
# D <0..255> — скважность прямоугольника (128 = 50 %)
|
||
# M <0..127> — амплитуда синуса (127 = максимум)
|
||
# A — однократно измерить размах на ADC0 -> "A <мВ>"
|
||
# SWEEP <f0> <f1> <n> <dwell_ms>
|
||
# — логарифмический свип; "P <Гц> <мВ>" + импульс на GP14
|
||
# STOP — прервать свип
|
||
#
|
||
# ─── СХЕМА СИНУСА (обязательна: ШИМ даёт меандр, а не синус) ───
|
||
# GP15 ──[R1]──┬──[R2]──┬──────► OUT (вход DUT)
|
||
# │ │
|
||
# [C1] [C2]
|
||
# │ │
|
||
# GND GND
|
||
# R1 = R2 = 1 кОм, C1 = C2 = 47 нФ -> f_c ≈ 3,4 кГц (для f_s >= 20 кГц);
|
||
# если фактическая f_s ниже — увеличить C (см. README: f_c ~= f_s/6).
|
||
# Входам DUT выше 3,3 В нужен делитель перед ADC0 (GP26).
|
||
|
||
import array
|
||
import math
|
||
import sys
|
||
import time
|
||
|
||
import analogio
|
||
import board
|
||
import digitalio
|
||
import pwmio
|
||
import supervisor
|
||
|
||
PIN_OUT = board.GP15 # выход сигнала (синус — через RC-фильтр)
|
||
PIN_TRIG = board.GP14 # синхроимпульс (для осциллографа)
|
||
PIN_ADC = board.GP26 # ADC0 = A0 — вход измерения (через делителем)
|
||
|
||
CARRIER_HZ = 125000 # несущая ШИМ — много выше f_s, биений нет
|
||
PWM_RANGE = 65535 # duty_cycle pwmio — 16 бит
|
||
AMP_MAX = 127 # максимальная амплитуда синуса
|
||
SW_SQ_MIN_HZ = 20 # ниже — «программный» прямоугольник в цикле
|
||
FS_INIT = 20000 # оценка f_s до калибровки
|
||
STEP_2P32 = 1 << 32 # аккумулятор фазы 32 бит
|
||
|
||
# ─────────────────── состояние генератора ─────────────────────
|
||
state = {
|
||
"freq": 1000, # Гц
|
||
"duty": 128, # 0..255
|
||
"amp": AMP_MAX, # 0..127
|
||
"mode": "Q", # 'Q' — прямоугольник, 'S' — синус
|
||
# поля свипа (активен, если "sweep": True)
|
||
"sweep": False,
|
||
"sweep_f0": 0.0,
|
||
"sweep_ratio": 1.0,
|
||
"sweep_idx": 0,
|
||
"sweep_n": 0,
|
||
"sweep_dwell": 0.05,
|
||
"sweep_t0": 0.0,
|
||
}
|
||
fs_eff = FS_INIT # фактическая частота выборок (из калибровки)
|
||
|
||
# Таблица синуса 256 точек, -127..127
|
||
LUT = array.array("b", [
|
||
int(127 * math.sin(2 * math.pi * i / 256))
|
||
for i in range(256)
|
||
])
|
||
|
||
# ─────────────────────── периферия ─────────────────────────────
|
||
pwm = pwmio.PWMOut(PIN_OUT, frequency=CARRIER_HZ, duty_cycle=0,
|
||
variable_frequency=True)
|
||
trig = digitalio.DigitalInOut(PIN_TRIG)
|
||
trig.direction = digitalio.Direction.OUTPUT
|
||
adc = analogio.AnalogIn(PIN_ADC)
|
||
|
||
|
||
# ─────────────────────── слой ШИМ ─────────────────────────────
|
||
def pwm_config(freq_hz):
|
||
"""Частота несущей (прямоугольник) — аппаратный pwmio."""
|
||
pwm.frequency = max(1, int(freq_hz))
|
||
|
||
|
||
def pwm_duty(duty):
|
||
"""Скважность 0..255 -> duty_cycle 0..65535."""
|
||
if duty < 0:
|
||
duty = 0
|
||
if duty > 255:
|
||
duty = 255
|
||
pwm.duty_cycle = duty * PWM_RANGE // 255
|
||
|
||
|
||
def pwm_set_square(freq_hz, duty):
|
||
pwm_config(freq_hz)
|
||
pwm_duty(duty)
|
||
|
||
|
||
# ─────────────────── DDS: одна выборка ─────────────────────────
|
||
def dds_tick(phase):
|
||
"""Шаг аккумулятора фазы + запись скважности. Возвращает новую фазу.
|
||
|
||
В режиме SINE — выборка синуса; в режиме Q с f < SW_SQ_MIN_HZ —
|
||
«программный» прямоугольник. В режиме Q с f >= SW_SQ_MIN_HZ ШИМ работает
|
||
аппаратно, функция только возвращает фазу.
|
||
"""
|
||
f = state["freq"]
|
||
step = f * STEP_2P32 // fs_eff
|
||
phase = (phase + step) & 0xFFFFFFFF
|
||
if state["mode"] == "S":
|
||
idx = phase >> 24
|
||
s = LUT[idx] * state["amp"] // 127
|
||
pwm_duty(128 + s)
|
||
elif f < SW_SQ_MIN_HZ:
|
||
pwm_duty(255 if (phase >> 24) < state["duty"] else 0)
|
||
return phase
|
||
|
||
|
||
# ───────────── измерение размаха (ADC), не останавливая DDS ────
|
||
def measure_amplitude_mv(phase, window_s=0.05):
|
||
"""Размах (пик-пик) в мВ за окно; DDS при этом продолжает тикать."""
|
||
vmin = 65535
|
||
vmax = 0
|
||
t0 = time.monotonic()
|
||
while time.monotonic() - t0 < window_s:
|
||
phase = dds_tick(phase)
|
||
v = adc.value
|
||
if v < vmin:
|
||
vmin = v
|
||
if v > vmax:
|
||
vmax = v
|
||
# ADC 12 бит в 16-битной шкале CircuitPython, Vref = 3,3 В
|
||
return (vmax - vmin) * 3300 // 65535, phase
|
||
|
||
|
||
# ─────────────────────────── команды ───────────────────────────
|
||
def cmd_f(a):
|
||
hz = min(200000, max(1, int(float(a))))
|
||
state["freq"] = hz
|
||
if state["mode"] == "Q":
|
||
pwm_set_square(hz, state["duty"])
|
||
print("OK F", hz)
|
||
|
||
|
||
def cmd_w(a):
|
||
state["mode"] = "S" if a.upper().startswith("S") else "Q"
|
||
if state["mode"] == "Q":
|
||
pwm_set_square(state["freq"], state["duty"])
|
||
else:
|
||
pwm_config(CARRIER_HZ) # выборки пишет dds_tick
|
||
print("OK W", state["mode"])
|
||
if state["mode"] == "S":
|
||
print("# sine:", state["freq"], "Hz")
|
||
|
||
|
||
def cmd_d(a):
|
||
state["duty"] = min(255, max(0, int(a)))
|
||
if state["mode"] == "Q":
|
||
pwm_set_square(state["freq"], state["duty"])
|
||
print("OK D", state["duty"])
|
||
|
||
|
||
def cmd_m(a):
|
||
state["amp"] = min(AMP_MAX, max(0, int(a)))
|
||
print("OK M", state["amp"])
|
||
|
||
|
||
def cmd_sf(a):
|
||
"""Частота синусоиды (с защитой от алиасинга: f <= f_s/2)."""
|
||
hz = min(fs_eff // 2, max(1, int(float(a))))
|
||
state["freq"] = hz
|
||
print("OK SF", hz)
|
||
|
||
|
||
def cmd_sweep_parts(a):
|
||
parts = a.split()
|
||
if len(parts) != 4:
|
||
print("ERR SWEEP <f0> <f1> <n> <dwell_ms>")
|
||
return
|
||
f0 = max(1.0, float(parts[0]))
|
||
f1 = float(parts[1])
|
||
if f1 <= f0:
|
||
f1 = f0 * 10
|
||
n = max(2, int(parts[2]))
|
||
dwell = max(0.01, int(parts[3]) / 1000.0)
|
||
ratio = (f1 / f0) ** (1.0 / (n - 1))
|
||
state["sweep"] = True
|
||
state["sweep_f0"] = f0
|
||
state["sweep_ratio"] = ratio
|
||
state["sweep_idx"] = 0
|
||
state["sweep_n"] = n
|
||
state["sweep_dwell"] = dwell
|
||
state["sweep_t0"] = 0.0
|
||
cmd_f(str(int(round(f0))))
|
||
print("OK SWEEP")
|
||
|
||
|
||
def trigger_pulse():
|
||
trig.value = True
|
||
time.sleep(0.00005)
|
||
trig.value = False
|
||
|
||
|
||
def handle(line, phase):
|
||
"""Выполнить команду. Возвращает (ok, phase). Любые ошибки — внутрь."""
|
||
line = line.strip()
|
||
if not line:
|
||
return False, phase
|
||
parts = line.split(None, 1) # позиционно: sep=None, maxsplit=1 (в CP нет kwargs!)
|
||
cmd = parts[0].upper()
|
||
a = parts[1] if len(parts) > 1 else ""
|
||
if cmd == "F":
|
||
cmd_f(a)
|
||
elif cmd == "SF":
|
||
cmd_sf(a)
|
||
elif cmd == "W":
|
||
cmd_w(a)
|
||
elif cmd == "D":
|
||
cmd_d(a)
|
||
elif cmd == "M":
|
||
cmd_m(a)
|
||
elif cmd == "A":
|
||
mv, phase = measure_amplitude_mv(phase)
|
||
print("A", mv)
|
||
elif cmd == "SWEEP":
|
||
cmd_sweep_parts(a)
|
||
elif cmd == "STOP":
|
||
state["sweep"] = False
|
||
print("OK STOP")
|
||
else:
|
||
print("ERR unknown:", line)
|
||
return True, phase
|
||
|
||
|
||
# ─────────────────── калибровка f_s (1 с) ──────────────────────
|
||
def calibrate_fs(phase):
|
||
"""Прогон DDS в течение 1 с и подсчёт фактической f_s."""
|
||
t0 = time.monotonic()
|
||
n = 0
|
||
while time.monotonic() - t0 < 1.0:
|
||
phase = dds_tick(phase)
|
||
n += 1
|
||
return n, phase
|
||
|
||
|
||
# ─────────────────────────── main ─────────────────────────────
|
||
def main():
|
||
global fs_eff
|
||
print("# DDS generator ready (CircuitPython). help: F/W/D/A/SWEEP/STOP")
|
||
phase = 0
|
||
try:
|
||
fs_eff, phase = calibrate_fs(phase)
|
||
except Exception as e:
|
||
print("! calibrate error:", e)
|
||
print("# sample rate:", fs_eff, "Hz")
|
||
|
||
buf = ""
|
||
while True:
|
||
try:
|
||
phase = dds_tick(phase)
|
||
|
||
# свип: точка готова, когда прошло dwell
|
||
if state["sweep"]:
|
||
now = time.monotonic()
|
||
if state["sweep_t0"] == 0.0 or now - state["sweep_t0"] >= state["sweep_dwell"]:
|
||
mv, phase = measure_amplitude_mv(phase)
|
||
print("P", int(round(state["freq"])), mv)
|
||
state["sweep_idx"] += 1
|
||
if state["sweep_idx"] >= state["sweep_n"]:
|
||
state["sweep"] = False
|
||
print("DONE")
|
||
cmd_f(str(int(round(state["sweep_f0"]))))
|
||
else:
|
||
f = state["sweep_f0"] * (state["sweep_ratio"] ** state["sweep_idx"])
|
||
cmd_f(str(int(round(f))))
|
||
trigger_pulse()
|
||
state["sweep_t0"] = time.monotonic()
|
||
|
||
# командная строка (неблокирующий опрос USB-консоли)
|
||
while supervisor.runtime.serial_bytes_available:
|
||
try:
|
||
c = sys.stdin.read(1)
|
||
except Exception as e:
|
||
print("! serial read error:", e)
|
||
break
|
||
if not c:
|
||
break
|
||
if c == "\n" or c == "\r":
|
||
if buf.strip():
|
||
print("->", buf.strip()) # эхо принятой строки
|
||
try:
|
||
_, phase = handle(buf, phase)
|
||
except Exception as e:
|
||
print("! command error:", e)
|
||
buf = ""
|
||
elif c != "\x00":
|
||
buf += c
|
||
|
||
except Exception as e:
|
||
# НЕ умираем в REPL: печатаем и продолжаем
|
||
print("! loop error:", e)
|
||
time.sleep(0.05)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|