diff --git a/pywellen/pywellen/pywellen.pyi b/pywellen/pywellen/pywellen.pyi index 2340623..9e6b146 100644 --- a/pywellen/pywellen/pywellen.pyi +++ b/pywellen/pywellen/pywellen.pyi @@ -61,6 +61,7 @@ class Signal: 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: ... + def width(self) -> int: ... class SignalChangeIter: def __iter__(self) -> SignalChangeIter: ... diff --git a/pywellen/pywellen/signal_builder.py b/pywellen/pywellen/signal_builder.py index 81fecb7..288c5e0 100644 --- a/pywellen/pywellen/signal_builder.py +++ b/pywellen/pywellen/signal_builder.py @@ -1,13 +1,13 @@ import abc -from typing import List, Optional -from pywellen import Signal +from typing import Dict, List, Optional, Union +from pywellen import Signal, Waveform, create_derived_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]: + def get_value_at_index(self, time_table_idx: int) -> Optional[Union[int, str]]: """ Returns the value at the index -- wellen interprets this as an unsigned width of size `self.width` @@ -22,6 +22,26 @@ def width(self): timetable index """ + def to_signal(self) -> Signal: + return create_derived_signal(self) + + +class PassThrough(SignalBuilder): + signal: Signal + + def __init__(self, signal: Signal, lsb: int, msb: int): + + self.signal = signal + + def get_all_signals(self) -> List[Signal]: + return [self.signal] + + def get_value_at_index(self, time_table_idx: int) -> Optional[Union[int, str]]: + return self.signal.value_at_idx(time_table_idx) + + def width(self): + return self.signal.width + class SlicedSignal(SignalBuilder): signal: Signal @@ -37,7 +57,7 @@ def __init__(self, signal: Signal, lsb: int, msb: int): def get_all_signals(self) -> List[Signal]: return [self.signal] - def get_value_at_index(self, time_table_idx: int) -> Optional[int]: + def get_value_at_index(self, time_table_idx: int) -> Optional[Union[int, str]]: current_value = self.signal.value_at_idx(time_table_idx) if isinstance(current_value, int): mask = (1 << (self.msb - self.lsb)) - 1 @@ -47,3 +67,6 @@ def get_value_at_index(self, time_table_idx: int) -> Optional[int]: def width(self): return self.msb - self.lsb + + +def get_signals(wave: Waveform) -> Dict[str, SignalBuilder]: ... diff --git a/pywellen/src/lib.rs b/pywellen/src/lib.rs index d91341d..68f639d 100644 --- a/pywellen/src/lib.rs +++ b/pywellen/src/lib.rs @@ -1,5 +1,5 @@ mod convert; -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc}; use convert::Mappable; use num_bigint::BigUint; @@ -23,7 +23,7 @@ impl PyErrExt for wellen::Result { } #[pymodule] -fn pywellen(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> { +pub fn pywellen(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; @@ -33,6 +33,31 @@ fn pywellen(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> { Ok(()) } +pub fn execute_get_signals( + script: &str, + fn_name: &str, + wave_path: String, +) -> PyResult> { + pyo3::append_to_inittab!(pywellen); + let val: PyResult> = Python::with_gil(|py| { + let activators = PyModule::from_code_bound(py, script, "signal_get.py", "signal_get")?; + let wave = Bound::new(py, Waveform::new(wave_path, true, true)?)?; + + let all_waves: HashMap = + activators.getattr(fn_name)?.call1((wave,))?.extract()?; + + Ok(all_waves) + }); + let val = val? + .into_iter() + .map(|(name, signal)| (name, signal.to_wellen_signal().unwrap())) + .fold(HashMap::new(), |mut mapper, val| { + mapper.insert(val.0, val.1); + mapper + }); + Ok(val) +} + #[pyclass] #[derive(Clone)] struct Hierarchy(pub(crate) Arc); @@ -301,6 +326,12 @@ impl Signal { } } +impl Signal { + fn to_wellen_signal(self) -> Option { + Arc::try_unwrap(self.signal).ok() + } +} + #[pyfunction] fn create_derived_signal(pyinterface: Bound<'_, PyAny>) -> PyResult { let all_signals: Vec = pyinterface @@ -333,6 +364,34 @@ fn create_derived_signal(pyinterface: Bound<'_, PyAny>) -> PyResult { }) } +pub fn create_wellen_signal(pyinterface: Bound<'_, PyAny>) -> PyResult { + let all_signals: Vec = 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(sig) +} + #[pyclass] /// Iterates across all changes -- the returned object is a tuple of (Time, Value) struct SignalChangeIter {