mirror of
https://github.com/m-labs/artiq.git
synced 2024-12-20 08:56:28 +08:00
Robert Jordens
01416bb0be
These are contributions of >= 30% or >= 20 lines (half-automated).
I hereby resubmit all my previous contributions to the ARTIQ project
under the following terms:
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
Closes #130
Signed-off-by: Robert Jordens <jordens@gmail.com>
134 lines
3.7 KiB
Python
134 lines
3.7 KiB
Python
# Copyright (C) 2014, 2015 M-Labs Limited
|
|
# Copyright (C) 2014, 2015 Robert Jordens <jordens@gmail.com>
|
|
|
|
from copy import copy
|
|
from math import cos, pi
|
|
|
|
from artiq.wavesynth.coefficients import discrete_compensate
|
|
|
|
|
|
class Spline:
|
|
def __init__(self):
|
|
self.c = [0.0]
|
|
|
|
def set_coefficients(self, c):
|
|
if not c:
|
|
c = [0.]
|
|
self.c = copy(c)
|
|
discrete_compensate(self.c)
|
|
|
|
def next(self):
|
|
r = self.c[0]
|
|
for i in range(len(self.c) - 1):
|
|
self.c[i] += self.c[i + 1]
|
|
return r
|
|
|
|
|
|
class SplinePhase:
|
|
def __init__(self):
|
|
self.c = [0.0]
|
|
self.c0 = 0.0
|
|
|
|
def set_coefficients(self, c):
|
|
if not c:
|
|
c = [0.]
|
|
self.c0 = c[0]
|
|
c1p = c[1:]
|
|
discrete_compensate(c1p)
|
|
self.c[1:] = c1p
|
|
|
|
def clear(self):
|
|
self.c[0] = 0.0
|
|
|
|
def next(self):
|
|
r = self.c[0]
|
|
for i in range(len(self.c) - 1):
|
|
self.c[i] += self.c[i + 1]
|
|
self.c[i] %= 1.0
|
|
return r + self.c0
|
|
|
|
|
|
class DDS:
|
|
def __init__(self):
|
|
self.amplitude = Spline()
|
|
self.phase = SplinePhase()
|
|
|
|
def next(self):
|
|
return self.amplitude.next()*cos(2*pi*self.phase.next())
|
|
|
|
|
|
class Wave:
|
|
def __init__(self):
|
|
self.bias = Spline()
|
|
self.dds = DDS()
|
|
self.v = 0.
|
|
self.silence = False
|
|
|
|
def next(self):
|
|
v = self.bias.next() + self.dds.next()
|
|
if not self.silence:
|
|
self.v = v
|
|
return self.v
|
|
|
|
def set_silence(self, s):
|
|
self.silence = s
|
|
|
|
|
|
class TriggerError(Exception):
|
|
pass
|
|
|
|
|
|
class Synthesizer:
|
|
def __init__(self, nchannels, program):
|
|
self.channels = [Wave() for _ in range(nchannels)]
|
|
self.program = program
|
|
# line_iter is None: "wait for segment selection" state
|
|
# otherwise: iterator on the current position in the frame
|
|
self.line_iter = None
|
|
|
|
def select(self, selection):
|
|
if self.line_iter is not None:
|
|
raise TriggerError("a frame is already selected")
|
|
self.line_iter = iter(self.program[selection])
|
|
self.line = next(self.line_iter)
|
|
|
|
def trigger(self):
|
|
if self.line_iter is None:
|
|
raise TriggerError("no frame selected")
|
|
|
|
line = self.line
|
|
if not line.get("trigger", False):
|
|
raise TriggerError("segment is not triggered")
|
|
|
|
r = [[] for _ in self.channels]
|
|
while True:
|
|
for channel, channel_data in zip(self.channels,
|
|
line["channel_data"]):
|
|
if "bias" in channel_data:
|
|
channel.bias.set_coefficients(
|
|
channel_data["bias"]["amplitude"])
|
|
if "dds" in channel_data:
|
|
channel.dds.amplitude.set_coefficients(
|
|
channel_data["dds"]["amplitude"])
|
|
if "phase" in channel_data["dds"]:
|
|
channel.dds.phase.set_coefficients(
|
|
channel_data["dds"]["phase"])
|
|
if channel_data["dds"].get("clear", False):
|
|
channel.dds.phase.clear()
|
|
channel.set_silence(channel_data.get("silence", False))
|
|
|
|
if line.get("dac_divider", 1) != 1:
|
|
raise NotImplementedError
|
|
|
|
for channel, rc in zip(self.channels, r):
|
|
for i in range(line["duration"]):
|
|
rc.append(channel.next())
|
|
|
|
try:
|
|
self.line = line = next(self.line_iter)
|
|
if line.get("trigger", False):
|
|
return r
|
|
except StopIteration:
|
|
self.line_iter = None
|
|
return r
|