Merge pull request #1035 from quartiq/fls-utils

Fls-utils

Author: jordens

py/pyproject.toml

@@ -18,7 +18,8 @@ dependencies = [
     "numpy",
     "scipy",
     "matplotlib",
-    "miniconf-mqtt@git+https://github.com/quartiq/miniconf@v0.19.0#subdirectory=py/miniconf-mqtt",
+    "tqdm",
+    "miniconf-mqtt@git+https://github.com/quartiq/miniconf@miniconf-v0.20.0#subdirectory=py/miniconf-mqtt",
 ]
 
 [project.urls]

py/stabilizer/search.py

@@ -0,0 +1,137 @@
+"""Stabilizer FLS signal searcher"""
+
+# pylint: disable=logging-fstring-interpolation,too-many-statements,too-many-locals,duplicate-code
+
+
+import asyncio
+import argparse
+import logging
+
+import numpy as np
+from tqdm.asyncio import tqdm
+
+import miniconf
+from miniconf.common import MQTTv5, one
+from stabilizer.stream import Stream, get_local_ip
+
+from . import stream as _
+
+_logger = logging.getLogger(__name__)
+
+
+async def main():
+    """FLS Signal searcher"""
+    parser = argparse.ArgumentParser(
+        description="Search demodulation frequency range and identify "
+        "highest power input frequency"
+    )
+    parser.add_argument(
+        "-v", "--verbose", action="count", default=0, help="Increase logging verbosity"
+    )
+    parser.add_argument(
+        "--broker",
+        "-b",
+        default="mqtt",
+        help="The MQTT broker address to use to use (%(default)s)",
+    )
+    parser.add_argument(
+        "--prefix",
+        "-p",
+        default="dt/sinara/fls/+",
+        help="The MQTT topic prefix (%(default)s)",
+    )
+    parser.add_argument(
+        "--channel",
+        "-c",
+        type=int,
+        choices=[0, 1],
+        default=0,
+        help="The channel to operate on",
+    )
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=9293,
+        help="Local port to listen on for streaming data",
+    )
+    parser.add_argument(
+        "--threshold",
+        "-t",
+        default=-15,
+        type=float,
+        help="log2 of tone power threshold",
+    )
+
+    args = parser.parse_args()
+    logging.basicConfig(
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        level=logging.WARN - 10 * args.verbose,
+    )
+
+    async with miniconf.Client(
+        args.broker,
+        protocol=MQTTv5,
+        logger=logging.getLogger("aiomqtt-client"),
+    ) as client:
+        prefix, _alive = one(await miniconf.discover(client, args.prefix))
+
+        conf = miniconf.Miniconf(client, prefix)
+        local_ip = get_local_ip(args.broker)
+        _transport, stream = await Stream.open(args.port, local_ip)
+
+        f2w = (1 << 32) / 500e6
+        f2s = 4 * 8 * 128 / 400e6
+        f_demod = 0x200000 / f2w
+
+        async def measure(f, nframe=1):
+            await conf.set(f"/ch/{args.channel}/input/freq", int(f * f2w))
+            # settle
+            # await asyncio.sleep(0.1)
+            # discard pending frames
+            while True:
+                try:
+                    stream.queue.get_nowait()
+                except asyncio.QueueEmpty:
+                    break
+            # discard one more
+            await stream.queue.get()
+            demod = []
+            for _ in range(nframe):
+                frame = await stream.queue.get()
+                demod.append(frame.demod()[:, args.channel])
+            demod = np.concatenate(demod)
+            p = np.log2(np.square(demod.astype(np.int64)).mean() / (1 << 62))
+            iq = demod.astype(np.float64).ravel().view(np.complex128)
+            df = np.angle(iq[1:] * iq[:-1].conj()).mean() / (2 * np.pi * f2s)
+            return f + df, p
+
+        try:
+            await conf.set("/stream", f"{local_ip}:{args.port}")
+            fp = []
+            with tqdm(np.arange(1e6, 240e6, 40e3)) as pbar:
+                async for f in pbar:
+                    ff, p = await measure(f)
+                    fp.append((ff, p))
+                    if p > args.threshold:
+                        _logger.info(f"{ff:g} Hz: {p:g} 3dB")
+                    else:
+                        _logger.debug(f"{ff:g} Hz: {p:g} 3dB")
+            if not fp:
+                raise ValueError("no tones found above threshold")
+            f0, _p0 = max(fp, key=lambda k: k[1])
+            fu, pu = await measure(f0 + 2 * f_demod)
+            _fl, pl = await measure(f0 - 2 * f_demod)
+            if pu > pl:
+                f0 = fu
+            f0, p0 = await measure(f0, nframe=100)
+            w = round(int(f0 * f2w))
+            _logger.warning(
+                f"final {f0:f} Hz, {p0:g} 3dB, ch/{args.channel}/input/freq={w:#x}"
+            )
+            await conf.set(f"/ch/{args.channel}/input/freq", w)
+        finally:
+            await conf.set("/stream", "0.0.0.0:0")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

py/stabilizer/stream.py

@@ -99,6 +99,29 @@ def to_si(self):
         )
 
 
+class Fls:
+    """FLS application stream format"""
+    format_id = 2
+
+    def __init__(self, header, body):
+        self.header = header
+        self.body = body
+
+    def size(self):
+        """Return the data size of the frame in bytes"""
+        return len(self.body)
+
+    def to_mu(self):
+        """Return the raw data in machine units"""
+        data = np.frombuffer(self.body, "<i4")
+        data = data.reshape(-1, 2, 7)
+        return data
+
+    def demod(self):
+        """Return baseband signal"""
+        return self.to_mu()[:, :, :2]
+
+
 class Frame:
     """Stream frame constisting of a header and multiple data batches"""
 
@@ -109,6 +132,7 @@ class Frame:
     parsers = {
         AdcDac.format_id: AdcDac,
         ThermostatEem.format_id: ThermostatEem,
+        Fls.format_id: Fls,
     }
 
     @classmethod

py/stabilizer/watch.py

@@ -0,0 +1,158 @@
+"""Stabilizer FLS event watcher and recorder"""
+
+# pylint: disable=logging-fstring-interpolation,too-many-statements
+
+import asyncio
+import argparse
+import logging
+import json
+import time
+
+import miniconf
+from miniconf.common import MQTTv5, one
+from stabilizer.stream import Frame, Stream, get_local_ip, wrap
+
+from . import stream as _
+
+_logger = logging.getLogger(__name__)
+
+
+async def main():
+    """FLS watcher"""
+    parser = argparse.ArgumentParser(
+        description="Watch telemetry messages for digital_input, blank, and slip "
+        "events and dump buffered stream data around those triggers into files."
+    )
+    parser.add_argument(
+        "-v", "--verbose", action="count", default=0, help="Increase logging verbosity"
+    )
+    parser.add_argument(
+        "--broker",
+        "-b",
+        default="mqtt",
+        help="The MQTT broker address to use to use (%(default)s)",
+    )
+    parser.add_argument(
+        "--prefix",
+        "-p",
+        default="dt/sinara/fls/+",
+        help="The MQTT topic prefix (%(default)s)",
+    )
+    parser.add_argument(
+        "--channel",
+        "-c",
+        type=int,
+        choices=[0, 1],
+        default=0,
+        help="The channel to operate on",
+    )
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=9293,
+        help="Local port to listen on for streaming data",
+    )
+    parser.add_argument(
+        "--buffer",
+        "-n",
+        default=1 << 12,
+        type=int,
+        help="number of packets to buffer (%(default)s)",
+    )
+    parser.add_argument(
+        "--dump", "-r", default="fls_event", help="file name prefix, (%(default)s)"
+    )
+    parser.add_argument(
+        "--max",
+        "-m",
+        default=20,
+        type=int,
+        help="max number of telemetry messages with active trigger"
+        " before exitin to limit the disk usage. (%(default)s)",
+    )
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        level=logging.WARN - 10 * args.verbose,
+    )
+
+    local_ip = get_local_ip(args.broker)
+
+    async def dump(ev):
+        transport, stream = await Stream.open(args.port, local_ip, maxsize=args.buffer)
+        try:
+            while True:
+                await ev.wait()
+                name = f"{args.dump}_{int(time.time())}.raw"
+                _logger.warning(f"trigger start {name}")
+                count = 0
+                with open(name, "wb") as f:
+                    while ev.is_set():
+                        fr = await stream.queue.get()
+                        f.write(Frame.header_fmt.pack(*fr.header))
+                        f.write(fr.body)
+                        count += 1
+                _logger.warning(f"trigger end, written {count} frames")
+        finally:
+            transport.close()
+
+    async def watch(ev):
+        async with miniconf.Client(
+            args.broker,
+            protocol=MQTTv5,
+            logger=logging.getLogger("aiomqtt-client"),
+        ) as client:
+            prefix, _alive = one(await miniconf.discover(client, args.prefix))
+            conf = miniconf.Miniconf(client, prefix)
+            await conf.set("/stream", f"{local_ip}:{args.port}")
+            await conf.close()
+
+            topic = f"{prefix}/telemetry"
+            await client.subscribe(topic)
+            try:
+                last = None
+                n = 0
+                async for tele in client.messages:
+                    if n > args.max:
+                        break
+                    tele = json.loads(tele.payload)
+                    if trigger(tele, last, args.channel):
+                        ev.set()
+                        n += 1
+                    else:
+                        ev.clear()
+                    last = tele
+            finally:
+                ev.clear()
+                await client.unsubscribe(topic)
+                conf = miniconf.Miniconf(client, prefix)
+                await conf.set("/stream", "0.0.0.0:0")
+                await conf.close()
+
+    def trigger(new, last, channel):
+        if last is None:
+            return False
+        if channel is None:
+            channels = [0, 1]
+        else:
+            channels = [channel]
+        for ch in channels:
+            if wrap(new["raw"][ch]["holds"] - last["raw"][ch]["holds"]) > 0:
+                continue
+            blanks = wrap(new["raw"][ch]["blanks"] - last["raw"][ch]["blanks"])
+            if blanks > 0:
+                _logger.info(f"trigger ch{ch} on {blanks} blanks")
+                return True
+            slips = wrap(new["raw"][ch]["slips"] - last["raw"][ch]["slips"])
+            if slips > 0:
+                _logger.info(f"trigger ch{ch} on {slips} slips")
+                return True
+        return False
+
+    ev = asyncio.Event()
+    await asyncio.gather(dump(ev), watch(ev))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/bin/dds.rs

@@ -254,7 +254,7 @@ mod app {
     }
 
     #[task(priority = 1, shared=[usb, settings], local=[usb_terminal])]
-    async fn usb(mut c: usb::Context) {
+    async fn usb(mut c: usb::Context) -> ! {
         loop {
             c.shared.usb.lock(|usb| {
                 usb.poll(&mut [c
@@ -275,7 +275,7 @@ mod app {
     }
 
     #[task(priority = 1, shared=[network])]
-    async fn ethernet_link(mut c: ethernet_link::Context) {
+    async fn ethernet_link(mut c: ethernet_link::Context) -> ! {
         loop {
             c.shared.network.lock(|net| net.processor.handle_link());
             Systick::delay(1.secs()).await;

src/bin/dual-iir.rs

@@ -497,7 +497,7 @@ mod app {
     }
 
     #[task(priority = 1, shared=[network, settings, telemetry], local=[cpu_temp_sensor])]
-    async fn telemetry(mut c: telemetry::Context) {
+    async fn telemetry(mut c: telemetry::Context) -> ! {
         loop {
             let telemetry =
                 c.shared.telemetry.lock(|telemetry| telemetry.clone());
@@ -526,7 +526,7 @@ mod app {
     }
 
     #[task(priority = 1, shared=[usb, settings], local=[usb_terminal])]
-    async fn usb(mut c: usb::Context) {
+    async fn usb(mut c: usb::Context) -> ! {
         loop {
             // Handle the USB serial terminal.
             c.shared.usb.lock(|usb| {
@@ -548,7 +548,7 @@ mod app {
     }
 
     #[task(priority = 1, shared=[network])]
-    async fn ethernet_link(mut c: ethernet_link::Context) {
+    async fn ethernet_link(mut c: ethernet_link::Context) -> ! {
         loop {
             c.shared.network.lock(|net| net.processor.handle_link());
             Systick::delay(1.secs()).await;