add signal builders interface

pywellen/pywellen/pywellen.pyi

@@ -1,10 +1,15 @@
 from typing import Optional, Tuple, Union
+from pywellen.signal_builder import SignalBuilder
 
 class Hierarchy:
     def all_vars(self) -> VarIter: ...
     def top_scopes(self) -> ScopeIter: ...
 
 class Scope:
+    """
+    One variable from a VCD/FST/GHW waveform
+    """
+
     def name(self, hier: Hierarchy) -> str: ...
     def full_name(self, hier: Hierarchy) -> str: ...
     def vars(self, hier: Hierarchy) -> VarIter: ...
@@ -15,6 +20,10 @@ class ScopeIter:
     def __next__(self) -> Scope: ...
 
 class Var:
+    """
+    One variable from a VCD/FST/GHW waveform
+    """
+
     def name(self, hier: Hierarchy) -> str: ...
     def full_name(self, hier: Hierarchy) -> str: ...
     def bitwidth(self) -> Optional[int]: ...
@@ -27,6 +36,10 @@ class TimeTable:
     def __getitem__(self, idx: int) -> int: ...
 
 class Waveform:
+    """
+    One VCD/FST/GHW waveform
+    """
+
     hierarchy: Hierarchy
     time_table: TimeTable
 
@@ -40,10 +53,17 @@ class Waveform:
     def get_signal_from_path(self, abs_hierarchy_path: str) -> Signal: ...
 
 class Signal:
+    """
+    Single signal in a VCD/GHW/FST
+
+    """
+
     def value_at_time(self, time: int) -> Union[int, str]: ...
     def value_at_idx(self, idx: int) -> Union[int, str]: ...
     def all_changes(self) -> SignalChangeIter: ...
 
 class SignalChangeIter:
     def __iter__(self) -> SignalChangeIter: ...
     def __next__(self) -> Tuple[int, str]: ...
+
+def create_derived_signal(builds: SignalBuilder) -> Signal: ...

pywellen/pywellen/signal_builder.py

@@ -0,0 +1,49 @@
+import abc
+from typing import List, Optional
+from pywellen import Signal
+
+
+class SignalBuilder(abc.ABC):
+    def get_all_signals(self) -> List[Signal]:
+        raise NotImplementedError()
+
+    def get_value_at_index(self, time_table_idx: int) -> Optional[int]:
+        """
+        Returns the value at the index -- wellen interprets this as an unsigned
+        width of size `self.width`
+        """
+        raise NotImplementedError()
+
+    def width(self):
+        """
+        Width of the generated signal in bits
+
+        It MUST be static -- width is not allowed to change function of
+        timetable index
+        """
+
+
+class SlicedSignal(SignalBuilder):
+    signal: Signal
+    lsb: int
+    msb: int
+
+    def __init__(self, signal: Signal, lsb: int, msb: int):
+
+        self.signal = signal
+        self.lsb = lsb
+        self.msb = msb
+
+    def get_all_signals(self) -> List[Signal]:
+        return [self.signal]
+
+    def get_value_at_index(self, time_table_idx: int) -> Optional[int]:
+        current_value = self.signal.value_at_idx(time_table_idx)
+        if isinstance(current_value, int):
+            mask = (1 << (self.msb - self.lsb)) - 1
+            return (current_value >> self.lsb) & mask
+        else:
+            return None
+
+    def width(self):
+        return self.msb - self.lsb

pywellen/src/convert.rs

@@ -12,6 +12,12 @@ pub trait Mappable: Sized {
     }
 }
 
+/// Trait to convert a data type into a `wellen::SignalValue`
+///
+pub trait ToValue: Sized {
+    fn as_signal_value(&self) -> SignalValue;
+}
+
 macro_rules! impl_mappable_basic {
     ($t:ty) => {
         impl Mappable for $t {

pywellen/src/lib.rs

@@ -9,7 +9,7 @@ use wellen::GetItem;
 
 use wellen::{
     viewers::{self},
-    LoadOptions, SignalValue, TimeTableIdx,
+    LoadOptions, TimeTableIdx,
 };
 
 pub trait PyErrExt<T> {
@@ -29,6 +29,7 @@ fn pywellen(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<Waveform>()?;
     m.add_class::<Signal>()?;
     m.add_class::<SignalChangeIter>()?;
+    m.add_function(wrap_pyfunction!(create_derived_signal, &m)?)?;
     Ok(())
 }
 
@@ -276,8 +277,8 @@ impl Signal {
             .map(|data_offset| self.signal.get_value_at(&data_offset, 0));
         if let Some(signal) = maybe_signal {
             let output = match signal {
-                SignalValue::Real(inner) => Some(inner.to_object(py)),
-                SignalValue::String(str) => Some(str.to_object(py)),
+                wellen::SignalValue::Real(inner) => Some(inner.to_object(py)),
+                wellen::SignalValue::String(str) => Some(str.to_object(py)),
                 _ => match BigUint::try_from_signal(signal) {
                     // If this signal is 2bits, this function will return an int
                     Some(number) => Some(number.to_object(py)),
@@ -300,6 +301,38 @@ impl Signal {
     }
 }
 
+#[pyfunction]
+fn create_derived_signal(pyinterface: Bound<'_, PyAny>) -> PyResult<Signal> {
+    let all_signals: Vec<Signal> = pyinterface
+        .call_method("get_all_signals", (), None)?
+        .extract()?;
+    let all_wellen_sigs = all_signals.iter().map(|sig| sig.signal.as_ref());
+    let all_changes = wellen::all_changes(all_wellen_sigs);
+    let bits: u32 = pyinterface.call_method("width", (), None)?.extract()?;
+    let mut builder = wellen::BitVectorBuilder::new(wellen::States::Two, bits);
+    for change_idx in all_changes {
+        let value: BigUint = pyinterface
+            .call_method("get_value_at_index", (change_idx,), None)?
+            .extract()?;
+        //FIXME: optimize this -- so much copying, needlessly
+        let bytes = value.to_bytes_be();
+        builder.add_change(
+            change_idx,
+            wellen::SignalValue::Binary(bytes.as_slice(), bits),
+        );
+    }
+    let sig = builder.finish(wellen::SignalRef::from_index(0xbeebabe5).unwrap());
+    let all_times = all_signals
+        .first()
+        .expect("No signals provided")
+        .all_times
+        .clone();
+    Ok(Signal {
+        signal: Arc::new(sig),
+        all_times,
+    })
+}
+
 #[pyclass]
 /// Iterates across all changes -- the returned object is a tuple of (Time, Value)
 struct SignalChangeIter {

pywellen/tests/test_waveform.py

@@ -1,6 +1,8 @@
 from pywellen import Waveform
-from pathlib import Path
+
 import subprocess
+from pywellen.pywellen import create_derived_signal
+from pywellen.signal_builder import SlicedSignal
 
 
 def _git_root_rel(path: str) -> str:
@@ -67,4 +69,40 @@ def test_vcd_not_starting_at_zero():
     assert sp_sig.value_at_idx(0) is None
 
 
-test_vcd_not_starting_at_zero()
+def test_slice():
+    filename = _git_root_rel("wellen/inputs/gameroy/trace_prefix.vcd")
+    waves = Waveform(path=filename)
+
+    h = waves.hierarchy
+
+    # the first signal change only happens at 4
+    assert waves.time_table[0] == 4
+
+    top = next(h.top_scopes())
+    assert top.name(h) == "gameroy"
+    cpu = next(top.scopes(h))
+
+    assert cpu.name(h) == "cpu"
+
+    pc = next(v for v in cpu.vars(h) if v.name(h) == "pc")
+    assert pc.full_name(h) == "gameroy.cpu.pc"
+    sp = next(v for v in cpu.vars(h) if v.name(h) == "sp")
+    assert sp.full_name(h) == "gameroy.cpu.sp"
+
+    ## querying a signal before it has a value should return none
+    pc_sig = waves.get_signal(pc)
+    sp_sig = waves.get_signal(sp)
+
+    ## pc is fine since it changes at 4 which is time_table idx 0
+    # pc_signal = waves.get_signal(pc.signal_ref())
+    assert pc_sig.value_at_idx(0) is not None
+    sliced_signal = create_derived_signal(SlicedSignal(signal=sp_sig, lsb=8, msb=16))
+
+    print(sp_sig.value_at_idx(1))
+    print(sliced_signal.value_at_idx(1))
+    ## sp only changes at 16 which is time table idx 1
+    assert sliced_signal.value_at_idx(1) is not None
+    assert sliced_signal.value_at_idx(0) is None
+
+
+test_slice()

wellen/src/derived_signals.rs

@@ -0,0 +1,116 @@
+use std::collections::HashMap;
+
+use crate::{
+    signal_utils::all_changes, BitVectorBuilder, Signal, SignalRef, SignalValue, States,
+    TimeTableIdx,
+};
+
+//TODO: support variable length signals -- currently only supports fixed length signals
+pub trait LazySignal {
+    fn value_at_idx(&self, time_idx: TimeTableIdx) -> Option<SignalValue>;
+    fn time_indices(&self) -> &[TimeTableIdx];
+    fn width(&self) -> u32;
+    fn states(&self) -> States;
+    fn to_signal(&self) -> Signal {
+        let mut vec_builder = BitVectorBuilder::new(self.states(), self.width());
+
+        for change in self.time_indices().iter().cloned() {
+            let value = self.value_at_idx(change);
+            if let Some(actual_val) = value {
+                vec_builder.add_change(change, actual_val)
+            }
+        }
+        vec_builder.finish(SignalRef::from_index(0xbeedad69).unwrap())
+    }
+}
+
+pub struct SimpleLazy {
+    signals: HashMap<String, Signal>,
+    generator: Box<dyn Fn(&'_ HashMap<String, Signal>, TimeTableIdx) -> Option<SignalValue<'_>>>,
+    all_times: Vec<TimeTableIdx>,
+    width: u32,
+}
+
+impl SimpleLazy {
+    pub fn new(
+        signals: HashMap<String, Signal>,
+        generator: Box<dyn Fn(&HashMap<String, Signal>, TimeTableIdx) -> Option<SignalValue>>,
+        width: u32,
+    ) -> SimpleLazy {
+        let all_times = all_changes(signals.values());
+        SimpleLazy {
+            all_times,
+            signals,
+            generator,
+            width,
+        }
+    }
+}
+
+impl LazySignal for SimpleLazy {
+    fn states(&self) -> States {
+        self.signals
+            .values()
+            .filter_map(|val| val.max_states())
+            .max()
+            .unwrap_or(States::Two)
+    }
+    fn value_at_idx(&self, time_idx: TimeTableIdx) -> Option<SignalValue> {
+        let gen = &self.generator;
+        gen(&self.signals, time_idx)
+    }
+    fn width(&self) -> u32 {
+        self.width
+    }
+    fn time_indices(&self) -> &[TimeTableIdx] {
+        &self.all_times
+    }
+}
+
+/// A signal that only emits values from a "source" signal when an "enable" signal changes
+///
+/// The enable signal doesnt _need_ to be a single bit -- a "clocking" event happens on any change
+pub struct VirtualEdgeTriggered<'a> {
+    source: &'a Signal,
+    virtual_enable: &'a Signal,
+    all_times: Vec<TimeTableIdx>,
+}
+
+impl<'a> VirtualEdgeTriggered<'a> {
+    pub fn new(source: &'a Signal, virtual_enable: &'a Signal) -> Option<Self> {
+        if source.width().is_some() && source.max_states().is_some() {
+            let all_times = all_changes(vec![source, virtual_enable]);
+            Some(Self {
+                source,
+                virtual_enable,
+                all_times,
+            })
+        } else {
+            None
+        }
+    }
+}
+impl<'a> LazySignal for VirtualEdgeTriggered<'a> {
+    fn value_at_idx(&self, time_idx: TimeTableIdx) -> Option<SignalValue> {
+        if let Some(offset) = self.virtual_enable.get_offset(time_idx) {
+            if offset.time_match {
+                self.source
+                    .get_offset(time_idx)
+                    .map(|offset| self.source.get_value_at(&offset, 0))
+            } else {
+                None
+            }
+        } else {
+            None
+        }
+    }
+    fn time_indices(&self) -> &[TimeTableIdx] {
+        self.all_times.as_slice()
+    }
+    fn width(&self) -> u32 {
+        self.source.width().unwrap()
+    }
+    fn states(&self) -> States {
+        self.source.max_states().unwrap()
+    }
+}

wellen/src/lib.rs

@@ -2,9 +2,11 @@
 // released under BSD 3-Clause License
 // author: Kevin Laeufer <[email protected]>
 
+pub mod derived_signals;
 mod fst;
 mod ghw;
 mod hierarchy;
+mod signal_utils;
 mod signals;
 pub mod simple;
 mod vcd;
@@ -57,7 +59,9 @@ pub use hierarchy::{
     GetItem, Hierarchy, HierarchyItem, Scope, ScopeRef, ScopeType, SignalEncoding, SignalRef,
     Timescale, TimescaleUnit, Var, VarDirection, VarIndex, VarRef, VarType,
 };
-pub use signals::{Real, Signal, SignalSource, SignalValue, Time, TimeTableIdx};
+pub use signal_utils::all_changes;
+pub use signals::{BitVectorBuilder, Real, Signal, SignalSource, SignalValue, Time, TimeTableIdx};
+pub use wavemem::States;
 
 #[cfg(feature = "benchmark")]
 pub use wavemem::check_states_pub;