forked from awendt/pyrelacs
51 lines
1.2 KiB
Python
51 lines
1.2 KiB
Python
import time
|
|
import numpy as np
|
|
from IPython import embed
|
|
from pyrelacs.dataio.circbuffer import CircBuffer
|
|
from pyrelacs.util.logging import config_logging
|
|
|
|
|
|
log = config_logging()
|
|
|
|
|
|
class SinProducer:
|
|
def __init__(
|
|
self,
|
|
buffer: CircBuffer,
|
|
) -> None:
|
|
self.buffer = buffer
|
|
self.stop = False
|
|
|
|
def produce_sin(
|
|
self,
|
|
*args,
|
|
**kwargs,
|
|
) -> None:
|
|
AMPLITUDE = 2
|
|
FREQUENCY = 10
|
|
|
|
self.stop = False
|
|
log.debug("producing Sin")
|
|
start_time = time.time()
|
|
t = 0
|
|
while not self.stop:
|
|
s = AMPLITUDE * np.sin(2 * np.pi * FREQUENCY * t)
|
|
self.buffer.append(s)
|
|
t += 1 / self.buffer.samplerate
|
|
time.sleep(1 / self.buffer.samplerate)
|
|
|
|
end_time = time.time()
|
|
|
|
log.debug(f"duration sinus {end_time-start_time}")
|
|
log.debug(f"Stimulation time {t}")
|
|
log.debug(f"Total samples produced {self.buffer.totalcount()}")
|
|
|
|
def stop_request(self):
|
|
self.stop = True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
buf = CircBuffer(1_000_000, 1, samplerate=10_000)
|
|
pro_sin = SinProducer(buf)
|
|
pro_sin.produce_sin()
|