TimeStepper¶

class photonforge.TimeStepper(**kwargs)¶

Abstract time stepper class used to time step a component.

Time steppers must be derived from this class and implement the folowing interface:

keys: attribute or property with a tuple of strings corresponding to the keys in TimeSeries.values (usually “port_name@mode_index”) in a fixed order. It is used to index into input and output arrays.
step_single(): Take a numpy array of complex inputs (in the order defined by keys), calculate the outputs for the current time step, and write them into a pre-allocated output array.

Time steppers may optionally override:

setup_state(), called with the Component, time step and index, and other arguments to initialize the time stepper.
reset(), used to reset any internal state the time stepper may hold. An override must call super().reset(), which restores the arrays given to bind_native() and the delay lines.
bind_native(), called from setup_state() to hand the simulator a compiled step function instead of implementing step_single(). Overriding step_single() in a subclass of a native stepper puts that subclass back on the Python path.
path_delays, set in setup_state(): the minimum delay, in time steps, of the signal path from each input key to each output key, as a dictionary keyed by (input, output). Missing pairs have no path. The default, None, assumes every input reaches every output within the time step, which is always correct but keeps the stepper inside any circuit loop it touches. Circuits schedule their instances from these declarations, so declare them exactly. Only whether a delay is zero is used today: a zero-delay path is resolved within the time step, a longer one from earlier samples. The value is a lower bound, which circuits sum into their own path_delays.
input_delay_steps and output_delay_steps, set in setup_state(): delays in time steps, keyed by port, that the simulator imposes around the stepper’s body instead of the body keeping them: step() and circuits read a delayed input from a line filled by earlier samples and hand a delayed output to its line, so a delayed element is called once per sample and its outputs are known ahead of the call. step_single() is the body alone. path_delays describes body and lines together.

Additionally, for time stepper parameterization to work correctly, the __init__ method must call super().__init__ with all parameters as keyword arguments (see example below).

Important

The following attributes and methods defined in the base class should never be redefined in the derived one: step(), setup(), update(), parametric_function, and parametric_kwargs.

Note

Time steppers are not thread-safe. Do not call step() or step_single() concurrently from multiple threads on the same instance.

Example

>>> class CustomTimeStepper(pf.TimeStepper):
...     def __init__(self, *, k=0.5):
...         super().__init__(k=k)
...
...     def setup_state(self, *, component, time_step, **kwargs):
...         self._k = self.parametric_kwargs["k"] ** time_step
...         ports = component.select_ports("optical")
...         self._keys = [f"{p}@0" for p in sorted(ports)]
...         self._state = np.zeros(len(self._keys), dtype=complex)
...
...     def reset(self):
...         self._state.fill(0)
...
...     @property
...     def keys(self):
...         return self._keys
...
...     def step_single(
...         self,
...         inputs,
...         outputs,
...         time_index,
...         update_state,
...         shutdown,
...     ):
...         np.copyto(outputs, self._state * self._k + inputs)
...         if update_state:
...             np.copyto(self._state, outputs)
...
>>> pf.register_time_stepper_class(CustomTimeStepper)
>>> time_stepper = CustomTimeStepper(k=0.2)

Methods

`bind_native`(step, keys[, state, cstate, context])	Bind a compiled step function.
`reset`()	Reset any internal state of the time stepper.
`setup`(component, time_step, *[, ...])	Initialize the time stepper.
`setup_state`(*, component, time_step, ...)	Initialize any internal state in preparation for one or more runs.
`step`([inputs, steps, time_step, show_progress])	Compute the outputs of this time stepper, given inputs.
`step_single`(inputs, outputs, time_index, ...)	Take a single time step on the given inputs.
`update`(args, *kwargs)	Update this time stepper.
`write_verilog_a`(path, *[, backend, ...])	Write this configured time stepper as one Verilog-A module.

Attributes

`parametric_function`	Function used to update the time stepper.
`parametric_kwargs`	Keyword arguments used to update the time stepper.
`path_delays`	Minimum delay in time steps of each (input key, output key) signal path; missing pairs have no path and None means every input reaches every output within the time step.
`properties`	Object properties.
`random_variables`	Random variables associated to the time stepper's parameters.

bind_native(step, keys, state=None, cstate=None, context=None)¶

Bind a compiled step function.

Instead of implementing step_single(), a time stepper can hand the simulator the address of a C function with the signature:

int64 step(inputs, outputs, time_index, update_state, shutdown,
           context, state, cstate)

inputs and outputs are complex128 arrays with one entry per key, state and cstate are the float64 and complex128 arrays, and a nonzero return signals failure. The function must overwrite every element of outputs and only modify the state arrays when update_state is true. The arrays must not be reallocated after binding; reset() restores them to the values given here, so a subclass overriding reset() must call super().reset().

Calls with update_state false must give the same outputs: circuits may call the function several times in a time step, and setup checks this for steppers that declare path_delays. Random numbers must therefore come from a generator whose state lives in the arrays (e.g., a seed and a counter): output a draw kept there and draw the next one only when update_state is true, which also makes reset() replay the stream. Do not use numpy.random in the function: numba draws it from a global stream that differs on every call and survives a reset.

The function can be compiled with numba (when available).

Parameters:

step (int | object) – Address of the step function, or an object with an address attribute such as a numba cfunc.
keys (Sequence[str]) – Input/output keys.
state (ndarray | None) – Contiguous float64 array the function may use.
cstate (ndarray | None) – Contiguous complex128 array the function may use.
context (int | None) – Opaque pointer passed to the function.

Example

A two-port element that leaks the field entering its first port out of its second one:

>>> from numba import cfunc
>>> from photonforge.time_steppers import STEP_SIGNATURE
>>>
>>> @cfunc(STEP_SIGNATURE, cache=True)  
... def leaky_step(inputs, outputs, time_index, update_state,
...                shutdown, context, state, cstate):
...     # state[0]: decay per step
...     # cstate[0]: field leaving the second port
...     outputs[0] = 0j
...     outputs[1] = cstate[0] * state[0] + inputs[0]
...     if update_state:
...         cstate[0] = outputs[1]
...     return 0
...
>>> class LeakyTimeStepper(pf.TimeStepper):
...     def __init__(self, *, k=0.5):
...         super().__init__(k=k)
...
...     def setup_state(self, *, component, time_step, **kwargs):
...         keys = tuple(f"{p}@0" for p in sorted(component.ports))
...         state = numpy.array([
...             self.parametric_kwargs["k"] ** time_step
...         ])
...         self.bind_native(
...             leaky_step, keys, state, numpy.zeros(1, complex)
...         )
...
>>> pf.register_time_stepper_class(LeakyTimeStepper)

parametric_function¶

Function used to update the time stepper.

Type:: str

parametric_kwargs¶

Keyword arguments used to update the time stepper.

Type:: dict[str, object]

path_delays¶

Minimum delay in time steps of each (input key, output key) signal path; missing pairs have no path and None means every input reaches every output within the time step.

Type:: dict[tuple[str, str], int] | None

properties¶

Object properties.

Type:: Properties

random_variables¶

Random variables associated to the time stepper’s parameters.

Type:: list[RandomVariable]

reset()¶

Reset any internal state of the time stepper.

The default implementation restores the arrays given to bind_native(), if any, and performs no other action. Derived classes override it to reset the state they hold themselves.

setup(component, time_step, *, carrier_frequency=0, show_progress=True, **kwargs)¶

Initialize the time stepper.

Parameters:

component (Component) – Component for the time stepper.
time_step (float) – The interval between time steps (in seconds).
carrier_frequency (float) – The carrier frequency used to construct the time stepper. The carrier should be omitted from the input signals, as it is handled automatically by the time stepper.
show_progress (bool) – If True, show setup progress.
**kwargs (object) – Keyword args forwarded to setup_state().

Returns:

This time stepper.

Return type:

TimeStepper

setup_state(*, component, time_step, carrier_frequency, show_progress, **kwargs)¶

Initialize any internal state in preparation for one or more runs.

The default implementation performs no action. This method can be overridden in derived classes to initialize any internal state the time stepper may have.

Long-running processes can immediately return an object with an status``attribute that provides a dictionary with keys ``"progress" (a number from 0 to 100 indicating the computation progress) and "message" (one of 'running', 'success', or 'error', indicating the current computation status).

Parameters:

component (Component) – Component for the time stepper.
time_step (float) – The interval between time steps (in seconds).
carrier_frequency (float) – The carrier frequency used to construct the time stepper. The carrier should be omitted from the input signals, as it is handled automatically by the time stepper.
show_progress (bool) – If True, show setup progress.
**kwargs (object) – Keyword arguments forwarded from setup().

Returns:

Optional object with a status dictionary.

Return type:

object

step(inputs=None, steps=0, time_step=0, show_progress=True)¶

Compute the outputs of this time stepper, given inputs.

Parameters:

inputs (TimeSeries) – Inputs to the time stepper.
steps (int) – The number of time steps to execute. If inputs is defined, this value is not used.
time_step (float) – The interval between time steps (in seconds). If inputs is a defined, this value is not used.
show_progress (bool) – If True, show computation progress.

Returns:

Computed outputs.

Return type:

TimeSeries

step_single(inputs, outputs, time_index, update_state, shutdown)¶

Take a single time step on the given inputs.

Important

Derived classes must implement this method unless they bind a compiled step with bind_native(), in which case the default implementation calls it and, when shutdown is set, the optional _after_native_run hook.

The method takes the inputs as a numpy array of type complex with values in the order defined by keys. It advances a single time step, updating its internal state (if any) according to the update_state flag. It writes the outputs in-place to the pre-allocated, zero-initialized outputs array in the same order.

Parameters:

inputs (ndarray) – Input values at the current time step. Must be a 1D array of complex values ordered according to keys.
outputs (ndarray) – Pre-allocated output array where results will be stored. Same size and type as inputs.
time_index (int) – Time series index for the current input.
update_state (bool) – Whether to update the internal stepper state.
shutdown (bool) – Whether this is the last call to the single stepping function for the original TimeSeries.

update(*args, **kwargs)¶

Update this time stepper.

All contents of the time stepper are updated with the contents from the updated version. Object contents or modifications introduced after the time stepper creation (outside of the original parametric function) are not propagated.

Parameters:

*args (object) – Positional arguments to the parametric time stepper function.
**kwargs (object) – Keyword arguments to the parametric time stepper function.

Returns:

This time stepper.

Return type:

TimeStepper

Note

The original keyword arguments used to generate the time stepper are stored and updated with the passed kwargs. This updated version is used in the function call and will be stored for future updates. Positional arguments are not stored.

write_verilog_a(path, *, backend='openvaf', port_convention='travelling_wave', module_name=None, z0=50.0, field_impedance=None)¶

Write this configured time stepper as one Verilog-A module.

The time stepper must first be configured with setup() or Component.setup_time_stepper(). The writer creates source only; it does not compile the model or create a simulator deck.

Parameters:

path (str | os.PathLike) – Destination .va file.
backend (Literal["openvaf", "spectre"]) – Compiler capability target recorded in the source.
port_convention (Literal["travelling_wave", "gf_45spclo_npe"]) – External optical port convention. GF packed ports require the Spectre backend.
module_name (str | None) – Module identifier derived from path when omitted.
z0 (float) – Travelling-wave optical reference impedance.
field_impedance (float | None) – GF electric-field scaling.

Returns:

Absolute destination path.

Return type:

pathlib.Path