API
examples
examples.devices
examples.devices.remote_controlled
- class examples.devices.remote_controlled.RemoteControlled(initial_observed: float = 0, initial_unobserved: float = 42, initial_hidden: float = 3.14)[source]
A trivial toy device which is controlled by an adapter.
A RemoteControlled constructor of the, which configures various initial values.
- Parameters
- Return type
- class Inputs(**kwargs)
An empty typed mapping of device inputs
- class Outputs(**kwargs)
A typed mapping containing the ‘observed’ output value
- update(time: SimTime, inputs: examples.devices.remote_controlled.Inputs) tickit.core.device.DeviceUpdate[examples.devices.remote_controlled.Outputs][source]
The update method which produces an output mapping containing the observed value.
- Parameters
- Returns
The produced update event which contains the observed value, the device never requests a callback.
- Return type
- class examples.devices.remote_controlled.RemoteControlledAdapter(device: examples.devices.remote_controlled.RemoteControlled, raise_interrupt: Callable[[], Awaitable[None]], server: tickit.core.adapter.ServerConfig)[source]
A trivial composed adapter which gets and sets device properties.
A constructor of the Shutter adapter, which builds the configured server.
- Parameters
device (Device) – The device which this adapter is attached to.
raise_interrupt (Callable) – A callback to request that the device is updated immediately.
server (ServerConfig) – The immutable data container used to configure a server.
- Return type
- on_connect() AsyncIterable[bytes][source]
An on_connect method which continiously sends the unobserved value to the client.
- Returns
An asynchronous iterable which regularly outputs the unobserved value.
- Return type
AsyncIterable[bytes]
- async get_observed_bytes() bytes[source]
A regex bytes command which returns the byte encoded value of observed.
- Returns
The big endian float encoded value of observed.
- Return type
- async get_observed_str() bytes[source]
A regex string command which returns the utf-8 encoded value of observed.
- Returns
The utf-8 encoded value of observed.
- Return type
- async set_observed_bytes(value: bytes) bytes[source]
A regex bytes command which sets and echos the value of observed.
- async set_observed_str(value: float) bytes[source]
A regex string command which sets and echos the value of observed.
- async get_unobserved_bytes() bytes[source]
A regex bytes command which returns the byte encoded value of unobserved.
- Returns
The big endian float encoded value of unobserved.
- Return type
- async get_unobserved_str() bytes[source]
A regex string command which returns the utf-8 encoded value of unobserved.
- Returns
The utf-8 encoded value of unobserved.
- Return type
- async set_unobserved_bytes(value: bytes) bytes[source]
A regex bytes command which sets and echos the value of unobserved.
- async set_unobserved_str(value: float) bytes[source]
A regex string command which sets and echos the value of unobserved.
- async misc() bytes[source]
A regex string command which returns a utf-8 encoded character.
- Returns
A utf-8 encoded character.
- Return type
A regex string command which sets the value of hidden.
A regex string command which returns nothing, hidden cannot be shown.
- Return type
- yield_observed(n: int = 10) AsyncIterable[bytes][source]
A regex string command which sends observed numerous times.
- class RemoteControlledAdapterConfig(server: tickit.core.adapter.ServerConfig)
A constructor of the Shutter adapter, which builds the configured server.
- Parameters
device (Device) – The device which this adapter is attached to.
raise_interrupt (Callable) – A callback to request that the device is updated immediately.
server (ServerConfig) – The immutable data container used to configure a server.
- Return type
examples.devices.shutter
- class examples.devices.shutter.Shutter(default_position: float, initial_position: Optional[float] = None)[source]
A toy device which downscales flux according to a set position.
A toy device which produces an output flux which is downscaled from the input flux according to an internal state position. The position may be altered by setting new a target which will be matched by position over a period of time determined by the rate.
A Shutter constructor which configures the initial and default position.
- Parameters
- Return type
- class Inputs(**kwargs)
A typed mapping containing the ‘flux’ input value
- class Outputs(**kwargs)
A typed mapping containing the ‘flux’ output value
- static move(position: float, target: float, rate: float, period: SimTime) float[source]
A helper method used to compute the new position of a shutter.
A helper method used to compute the new position of a shutter given a target position, a rate of change and a period over which the change occurs. Movement is performed at the defined rate and comes to a “hard” stop when the desired position is reached.
- update(time: SimTime, inputs: examples.devices.shutter.Inputs) tickit.core.device.DeviceUpdate[examples.devices.shutter.Outputs][source]
The update method which moves the shutter and produces a downscaled flux.
The update method which adjusts the position according to the target position, computes the transmitted flux and produces the output flux with a request to be called back in 100ms if the if the shutter continues to move.
- Parameters
- Returns
The produced update event which contains the value of the transmitted flux, and requests a callback after 100ms if the shutter continues to move.
- Return type
- class examples.devices.shutter.ShutterAdapter(device: examples.devices.shutter.Shutter, raise_interrupt: Callable[[], Awaitable[None]], host: str = 'localhost', port: int = 25565)[source]
A toy composed adapter which gets shutter position and target and sets target.
A Shutter which instantiates a TcpServer with configured host and port.
- Parameters
device (Device) – The device which this adapter is attached to
raise_interrupt (Callable) – A callback to request that the device is updated immediately.
host (Optional[str]) – The host address of the TcpServer. Defaults to “localhost”.
port (Optional[int]) – The bound port of the TcpServer. Defaults to 25565.
- Return type
- async get_position() bytes[source]
A regex string command which returns the utf-8 encoded value of position.
- Returns
The utf-8 encoded value of position.
- Return type
- async get_target() bytes[source]
A regex string command which returns the utf-8 encoded value of target.
- Returns
The utf-8 encoded value of target.
- Return type
- async set_target(target: str) None[source]
A regex string command which sets the target position of the shutter.
- class ShutterAdapterConfig(host: str = 'localhost', port: int = 25565)
A Shutter which instantiates a TcpServer with configured host and port.
- Parameters
device (Device) – The device which this adapter is attached to
raise_interrupt (Callable) – A callback to request that the device is updated immediately.
host (Optional[str]) – The host address of the TcpServer. Defaults to “localhost”.
port (Optional[int]) – The bound port of the TcpServer. Defaults to 25565.
- Return type
examples.devices.trampoline
- class examples.devices.trampoline.RandomTrampoline(callback_period: int = 1000000000)[source]
A trivial toy device which produced a random output and requests a callback.
A constructor of the sink which configures the device callback period.
- Parameters
callback_period (int) – The simulation time callback period of the device (in nanoseconds). Defaults to int(1e9).
- Return type
- class Inputs(**kwargs)
- class Outputs(**kwargs)
- update(time: SimTime, inputs: examples.devices.trampoline.Inputs) tickit.core.device.DeviceUpdate[examples.devices.trampoline.Outputs][source]
The update method which produces a random output and requests a callback.
The update method which prints the time of the update, the inputs and the output which will be produced then returns the random output value and a request to be called back after the configured callback period.
- Parameters
- Returns
The produced update event which contains the value of the random output, and requests a callback after the configured callback period.
- Return type
- class examples.devices.trampoline.Trampoline(callback_period: int = 1000000000)[source]
A trivial toy device which requests a callback every update.
A constructor of the sink which configures the device callback period.
- Parameters
callback_period (int) – The simulation time callback period of the device (in nanoseconds). Defaults to int(1e9).
- Return type
- class Inputs(**kwargs)
An empty typed mapping of device inputs
- class Outputs(**kwargs)
An empty typed mapping of device outputs
- update(time: SimTime, inputs: examples.devices.trampoline.Inputs) tickit.core.device.DeviceUpdate[examples.devices.trampoline.Outputs][source]
The update method which prints the inputs and requests a callback.
The update method which prints the time of the update and the inputs then returns an empty output mapping and a request to be called back after the configured callback period.
- Parameters
- Returns
The produced update event which never contains any changes, and requests a callback after the configured callback period.
- Return type
tickit
- tickit.__version__: str
Version number as calculated by https://github.com/dls-controls/versiongit
tickit.adapters
tickit.adapters.interpreters
tickit.adapters.interpreters.command
tickit.adapters.interpreters.command.command_interpreter
- class tickit.adapters.interpreters.command.command_interpreter.Command(*args, **kwds)[source]
An interface for interperable commands.
- abstract parse(data: bytes) Optional[Sequence][source]
An abstract method which parses a message and extracts arguments.
An abstract method which parses a message and produces arguments if a match is found, otherwise None is returned.
- Parameters
data (bytes) – The message to be parsed.
- Returns
A sequence of arguments extracted from the message if matched, otherwise None.
- Return type
Optional[Sequence[AnyStr]]
- class tickit.adapters.interpreters.command.command_interpreter.CommandInterpreter[source]
An interpreter which routes to commands registered to adapter methods.
An interpreter which attempts to parse messages according to the parse method of commands registered against adapter methods, if a match is found the method is called with the parsed arguments.
- static unknown_command() AsyncIterable[bytes][source]
An asynchronous iterable of containing a single unknown command reply.
- Returns
An asynchronous iterable of containing a single unknown command reply: “Request does not match any known command”.
- Return type
AsyncIterable[bytes]
- async handle(adapter: tickit.core.adapter.Adapter, message: bytes) Tuple[AsyncIterable[Union[str, bytes]], bool][source]
Matches the message to an adapter command and calls the corresponding method.
An asynchronous method which handles a message by attempting to match the message against each of the registered commands, if a match is found the corresponding command is called and its reply is returned with an asynchronous iterable wrapper if required. If no match is found the unknown command message is returned with no request for interrupt.
- Parameters
- Returns
A tuple of the asynchronous iterable of reply messages and a flag indicating whether an interrupt should be raised by the adapter.
- Return type
tickit.adapters.interpreters.command.regex_command
- class tickit.adapters.interpreters.command.regex_command.RegexCommand(regex: AnyStr, interrupt: bool = False, format: Optional[str] = None)[source]
A decorator to register an adapter method as a regex parsed command.
- Parameters
regex (Union[bytes, str]) – The regular expression pattern which must be matched in full, with groups used to extract command arguments.
interrupt (bool) – A flag indicating whether calling of the method should raise an adapter interrupt. Defaults to False.
format (Optional[str]) – The message decoding format to be used for string based interpretation. Defaults to None.
- Returns
A decorator which registers the adapter method as a message handler.
- Return type
Callable
- parse(data: bytes) Optional[Sequence][source]
Performs message decoding and regex matching to match and extract arguments.
A method which performs message decoding accoridng to the command formatting string, checks for a full regular expression match and returns a sequence of function arguments if a match is found, otherwise the method returns None.
- Parameters
data (bytes) – The message data to be parsed.
- Returns
If a full match is found a sequence of function arguments is returned, otherwise the method returns None.
- Return type
Optional[Sequence[AnyStr]]
tickit.adapters.servers
tickit.adapters.servers.tcp
- class tickit.adapters.servers.tcp.TcpServer(host: str = 'localhost', port: int = 25565, format: tickit.utils.byte_format.ByteFormat = ByteFormat(format=b'%b'))[source]
A configurable tcp server with delegated message handling for use in adapters.
The TcpServer constructor which takes a host, port and format byte string.
- Parameters
host (str) – The host name which the server should be run under.
port (int) – The port number which the server should listen to.
format (ByteFormat) – A formatting string for messages sent by the server, allowing for the prepending and appending of data. Defaults to b”%b”.
- Return type
- async run_forever(on_connect: Callable[[], AsyncIterable[bytes]], handler: Callable[[bytes], Awaitable[AsyncIterable[bytes]]]) None[source]
Runs the TCP server indefinitely on the configured host and port.
An asynchronous method used to run the server indefinitely on the configured host and port. Upon client connection, messages from the on_connect iterable will be sent. Upon recieving a message the server will delegate handling of it to the handler. Replies will be formatted according to the configured format string.
- class TcpServerConfig(host: str = 'localhost', port: int = 25565, format: tickit.utils.byte_format.ByteFormat = ByteFormat(format=b'%b'))
The TcpServer constructor which takes a host, port and format byte string.
- Parameters
host (str) – The host name which the server should be run under.
port (int) – The port number which the server should listen to.
format (ByteFormat) – A formatting string for messages sent by the server, allowing for the prepending and appending of data. Defaults to b”%b”.
- Return type
tickit.adapters.composed
- tickit.adapters.composed.T
Message type
alias of TypeVar(‘T’)
- class tickit.adapters.composed.ComposedAdapter(_device: tickit.core.device.Device, _raise_interrupt: Callable[[], Awaitable[None]], _server: tickit.core.adapter.Server, _interpreter: tickit.core.adapter.Interpreter)[source]
An adapter implementation which delegates to a server and interpreter.
An adapter implementation which delegates the hosting of an external messaging protocol to a server and message handling to an interpreter.
- Parameters
_device (tickit.core.device.Device) –
_raise_interrupt (Callable[[], Awaitable[None]]) –
_server (tickit.core.adapter.Server) –
_interpreter (tickit.core.adapter.Interpreter) –
- Return type
- on_connect() AsyncIterable[Optional[tickit.adapters.composed.T]][source]
An overridable asynchronous iterable which yields messages on client connection.
- Returns
An asynchronous iterable of messages.
- Return type
AsyncIterable[Optional[T]]
- async handle_message(message: tickit.adapters.composed.T) AsyncIterable[Optional[tickit.adapters.composed.T]][source]
Delegates message handling to the interpreter and raises interrupt if requested.
tickit.core
tickit.core.components
tickit.core.components.component
- class tickit.core.components.component.Component(*args, **kwds)[source]
An interface for types which implement stand-alone simulation components.
An interface for types which implement stand-alone simulation components. Components define the top level building blocks of a tickit simulation (examples include the DeviceSimulation which host a device and corresponding adapter or a SystemSimulation which hosts a SlaveScheduler and internal Components).
- class tickit.core.components.component.ComponentConfig(name: ComponentID, inputs: Dict[PortID, tickit.core.typedefs.ComponentPort])[source]
A data container for component configuration.
A data container for component configuration which acts as a named union of subclasses to facilitate automatic deserialization.
- Parameters
name (ComponentID) –
inputs (Dict[PortID, tickit.core.typedefs.ComponentPort]) –
- Return type
- static configures() Type[tickit.core.components.component.Component][source]
The Component class configured by this config type.
- Returns
The Component class configured by this config type.
- Return type
Type[Component]
- class tickit.core.components.component.ConfigurableComponent[source]
A mixin used to create a component with a configuration data container.
- class tickit.core.components.component.BaseComponent(name: ComponentID, state_consumer: Type[tickit.core.state_interfaces.state_interface.StateConsumer], state_producer: Type[tickit.core.state_interfaces.state_interface.StateProducer])[source]
A base class for compnents, implementing state interface related methods.
A BaseComponent constructor which stores state interface types.
- Parameters
name (ComponentID) – The unique identifier of the component.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
- Return type
- async handle_input(input: tickit.core.typedefs.Input)[source]
Calls on_tick when an input is recieved.
- Parameters
input (Input) – An immutable data container for Component inputs.
- async output(time: SimTime, changes: Changes, call_at: Optional[SimTime]) None[source]
Constructs and sends an Output message to the component output topic.
An asynchronous method which constructs an Output message tagged with the component name and sends it to the output topic of this component.
- async raise_interrupt() None[source]
Sends an Interrupt message to the component output topic.
An asynchronous method whicb constructs an Interrupt message tagged with the component name and sends it to the output topic of this component.
- Return type
- async set_up_state_interfaces()[source]
Creates and configures a state consumer and state producer.
An asynchronous method which creates a state consumer which is subscribed to the input topic of the component and calls back to handle_input, and a state producer to produce Interrupt or Output messages.
- abstract async on_tick(time: SimTime, changes: Changes)[source]
An abstract asynchronous method which implements the core logic of the component.
- class BaseComponentConfig(name: ComponentID, inputs: Dict[PortID, tickit.core.typedefs.ComponentPort])
A BaseComponent constructor which stores state interface types.
- Parameters
name (ComponentID) – The unique identifier of the component.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
inputs (Dict[PortID, tickit.core.typedefs.ComponentPort]) –
- Return type
- tickit.core.components.component.create_components(configs: Iterable[tickit.core.components.component.ComponentConfig], state_consumer: Type[tickit.core.state_interfaces.state_interface.StateConsumer], state_producer: Type[tickit.core.state_interfaces.state_interface.StateProducer]) List[tickit.core.components.component.Component][source]
Creates a list of components from component config objects.
- Parameters
configs (Iterable[ComponentConfig]) – An iterable of component configuration data containers.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the components.
state_producer (Type[StateProducer]) – The state producer class to be used by the components.
- Returns
A list of instantiated components.
- Return type
List[Component]
tickit.core.components.device_simulation
- class tickit.core.components.device_simulation.DeviceSimulation(name: ComponentID, state_consumer: Type[tickit.core.state_interfaces.state_interface.StateConsumer], state_producer: Type[tickit.core.state_interfaces.state_interface.StateProducer], device: tickit.core.device.DeviceConfig, adapters: List[tickit.core.adapter.AdapterConfig])[source]
A component containing a device and the corresponding adapters.
A component which thinly wraps a device and the corresponding adapters, this component delegates core behaviour to the update method of the device, whilst allowing adapters to raise interrupts.
A DeviceSimulation constructor which builds a device and adapters from config.
- Parameters
name (ComponentID) – The unique identifier of the device simulation.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
device (DeviceConfig) – An immuatable device configuration data container, used to construct the device.
adapters (List[AdapterConfig]) – A list of immutable adapter configuration data containers, used to construct adapters.
- async on_tick(time: SimTime, changes: Changes) None[source]
Delegates core behaviour to the device and calls adapter on_update.
An asynchronous method which updates device inputs according to external changes, delegates core behaviour to the device update method, informs ListeningAdapters of the update, computes changes to the state of the component and sends the resulting Output.
- class DeviceSimulationConfig(name: ComponentID, inputs: Dict[PortID, tickit.core.typedefs.ComponentPort], device: tickit.core.device.DeviceConfig, adapters: List[tickit.core.adapter.AdapterConfig])
A DeviceSimulation constructor which builds a device and adapters from config.
- Parameters
name (ComponentID) – The unique identifier of the device simulation.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
device (DeviceConfig) – An immuatable device configuration data container, used to construct the device.
adapters (List[AdapterConfig]) – A list of immutable adapter configuration data containers, used to construct adapters.
inputs (Dict[PortID, tickit.core.typedefs.ComponentPort]) –
- Return type
tickit.core.components.system_simulation
- class tickit.core.components.system_simulation.SystemSimulation(name: ComponentID, components: List[tickit.core.components.component.ComponentConfig], state_consumer: Type[tickit.core.state_interfaces.state_interface.StateConsumer], state_producer: Type[tickit.core.state_interfaces.state_interface.StateProducer], expose: Dict[PortID, tickit.core.typedefs.ComponentPort])[source]
A component containing a slave scheduler and several components.
A component which acts as a nested tickit simulation by wrapping a slave scheduler and a set of internal components, this component delegates core behaviour to the components within it, whilst outputting their requests for wakeups and interrupts.
A constructor which creates component simulations and adds exposing wiring.
- Parameters
name (ComponentID) – The unique identifier of the system simulation.
components (List[ComponentConfig]) – A list of immutable component configuration data containers, used to construct internal components.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
expose (Dict[PortID, ComponentPort]) – A mapping of outputs which the system simulation exposes and the corresponding output of an internal component.
- Return type
- async run_forever()[source]
Sets up state interfaces, the scheduler and components and blocks until any complete.
An asynchronous method starts the run_forever method of each component, runs the scheduler, and sets up externally facing state interfaces. The method blocks until and of the components or the scheduler complete.
- async on_tick(time: SimTime, changes: Changes) None[source]
Delegates core behaviour to the slave scheduler.
An asynchronous method which delegates core behaviour of computing changes and determining a callback period to the slave shceduler and sends the resulting Output.
- class SystemSimulationConfig(name: ComponentID, inputs: Dict[PortID, tickit.core.typedefs.ComponentPort], components: List[tickit.core.components.component.ComponentConfig], expose: Dict[PortID, tickit.core.typedefs.ComponentPort])
A constructor which creates component simulations and adds exposing wiring.
- Parameters
name (ComponentID) – The unique identifier of the system simulation.
components (List[ComponentConfig]) – A list of immutable component configuration data containers, used to construct internal components.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
expose (Dict[PortID, ComponentPort]) – A mapping of outputs which the system simulation exposes and the corresponding output of an internal component.
inputs (Dict[PortID, tickit.core.typedefs.ComponentPort]) –
- Return type
tickit.core.management
tickit.core.management.schedulers
tickit.core.management.schedulers.base
- class tickit.core.management.schedulers.base.BaseScheduler(wiring: Union[tickit.core.management.event_router.Wiring, tickit.core.management.event_router.InverseWiring], state_consumer: Type[tickit.core.state_interfaces.state_interface.StateConsumer], state_producer: Type[tickit.core.state_interfaces.state_interface.StateProducer])[source]
A base scheduler class which implements logic common to all schedulers.
A constructor which stores inputs and creates an empty wakeups mapping.
- Parameters
wiring (Union[Wiring, InverseWiring]) – A wiring or inverse wiring object representing the connections between components in the system.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
- Return type
- abstract async schedule_interrupt(source: ComponentID) None[source]
An abstract asynchronous method which should schedule an interrupt immediately.
- Parameters
source (ComponentID) – The component which should be updated.
- Return type
- async update_component(input: tickit.core.typedefs.Input) None[source]
Sends a message with an input to the input topic of a component.
- async handle_message(message: Union[tickit.core.typedefs.Interrupt, tickit.core.typedefs.Output]) None[source]
A callback to handle interrupts or outputs produced by the state consumer.
An asynchronous callback which handles interrupt and output messages produced by the state consumer; For Outputs, changes are propagated and wakeups scheduled if required, whilst handling of interrupts is deferred.
- async setup() None[source]
Instantiates and configures the ticker and state interfaces.
An asynchronous setup method which creates a ticker, a state consumer which is subscribed to the output topics of each component in the system, a state producer to produce component inputs.
- Return type
- add_wakeup(component: ComponentID, when: SimTime) None[source]
Adds a wakeup to the mapping.
- Parameters
component (ComponentID) – The component which should be updated.
when (SimTime) – The simulation time at which the update should occur.
- Return type
- get_first_wakeups() Tuple[Set[ComponentID], Optional[SimTime]][source]
Gets the components which are due for update first and the wakeup time.
A method which returns a set of components which are due for update first and the simulation time at which the updates are scheduled. Or an empty set and None if no wakeups are scheduled.
- Returns
- A tuple containing the set of
components which are scheduled for the first wakeup and the time at which the wakeup should occur. Or an empty set and None if no wakeups are scheduled.
- Return type
Tuple[Set[ComponentID], Optional[SimTime]]
tickit.core.management.schedulers.master
- class tickit.core.management.schedulers.master.MasterScheduler(wiring: Union[tickit.core.management.event_router.Wiring, tickit.core.management.event_router.InverseWiring], state_consumer: Type[tickit.core.state_interfaces.state_interface.StateConsumer], state_producer: Type[tickit.core.state_interfaces.state_interface.StateProducer], initial_time: int = 0, simulation_speed: float = 1.0)[source]
A master scheduler which orchestrates the running of a tickit simulation.
A master sheduler constructor which stores values for reference.
- Parameters
wiring (Union[Wiring, InverseWiring]) – A wiring or inverse wiring object representing the connections between components in the system.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
initial_time (int) – The initial time of the simulation (in nanoseconds). Defaults to 0.
simulation_speed (float) – The rate at which simulation time progresses with respect to real world time. Defaults to 1.0.
- Return type
- async setup() None[source]
Performs base setup and creates an awaitable flag to indicate new wakeups.
- Return type
- add_wakeup(component: ComponentID, when: SimTime) None[source]
Adds a wakeup to the priority queue and sets an awaitable flag.
A method which adds a wakeup to the priority queue and sets a awaitable flag indicating that a new wakeup has been added.
- Parameters
component (ComponentID) – The component which should be updated.
when (SimTime) – The simulation time at which the update should occur.
- Return type
- async run_forever() None[source]
Performs an intial tick then continiously schedules ticks according to wakeups.
An asynchronous method which initially performs setup and an initial tick in which all components are updated, subsequently ticks are performed as requested by components of the simulation according to the simulation speed.
- Return type
- async schedule_interrupt(source: ComponentID) None[source]
Schedules the interrupt of a component immediately.
An asynchronous method which implements the superclass abstract, an interrupt is sheduled as an immediate wakeup. This is achieved by giving the wakeup a simulation time equal to the simulation time of the last tick plus the real world time which has passed, adjusted by the simulation speed.
- Parameters
source (ComponentID) – The source component which should be updated.
- Return type
tickit.core.management.schedulers.slave
- class tickit.core.management.schedulers.slave.SlaveScheduler(wiring: Union[tickit.core.management.event_router.Wiring, tickit.core.management.event_router.InverseWiring], state_consumer: Type[tickit.core.state_interfaces.state_interface.StateConsumer], state_producer: Type[tickit.core.state_interfaces.state_interface.StateProducer], expose: Dict[PortID, tickit.core.typedefs.ComponentPort], raise_interrupt: Callable[[], Awaitable[None]])[source]
A slave scheduler which orchestrates nested tickit simulations.
A slave scheduler constructor which adds wiring and saves values for reference.
- Parameters
wiring (Union[Wiring, InverseWiring]) – A wiring or inverse wiring object representing the connections between components in the system.
state_consumer (Type[StateConsumer]) – The state consumer class to be used by the component.
state_producer (Type[StateProducer]) – The state producer class to be used by the component.
expose (Dict[PortID, ComponentPort]) – A mapping of slave scheduler outputs to internal component ports.
raise_interrupt (Callable[[], Awaitable[None]]) – A callback to request that the slave scheduler is updated immediately.
- Return type
- static add_exposing_wiring(wiring: Union[tickit.core.management.event_router.Wiring, tickit.core.management.event_router.InverseWiring], expose: Dict[PortID, tickit.core.typedefs.ComponentPort]) tickit.core.management.event_router.InverseWiring[source]
Adds wiring to expose slave scheduler outputs.
Adds wiring to expose slave scheduler outputs, this is performed creating a mock “expose” component with inverse wiring set by expose.
- Parameters
wiring (Union[Wiring, InverseWiring]) – A wiring or inverse wiring object representing the connections between components in the system.
expose (Dict[PortID, ComponentPort]) – A mapping of slave scheduler outputs to internal component ports.
- Returns
An inverse wiring object representing the connections between components in the system and the “expose” component which acts as the slave scheduler output.
- Return type
- async update_component(input: tickit.core.typedefs.Input) None[source]
Sends an input to the corresponding component. Mocks I/O for “external” or “expose”.
For real components the input is sent in a message to their input topic, for the mock component named “external”, external inputs are injected, whilst for the mock component and named “expose” the input is stored for use as the scheduler output.
- async on_tick(time: SimTime, changes: Changes) Tuple[Changes, Optional[SimTime]][source]
Routes inputs, performs a tick and returns output changes and a callback time.
An asyhcnronous method which determines which components within the simulation require being woken up, sets the input changes for use by the “external” mock component, performs a tick, determines the period in which the slave scheduler should next be updated, and returns the changes collated by the “expose” mock component.
- Parameters
- Returns
A tuple of a mapping of the changed exposed outputs and their new values and optionally a duration in simulation time after which the slave scheduler should be called again.
- Return type
- async run_forever() None[source]
Delegates to setup which instantiates the ticker and state interfaces.
- Return type
- async schedule_interrupt(source: ComponentID) None[source]
Schedules the interrupt of a component immediately.
An asynchronous method which schedules an interrupt immediately by adding it to a set of queued interrupts and raising the interrupt to the master scheduler.
- Parameters
source (ComponentID) – The source component of the interrupt.
- Return type
tickit.core.management.event_router
- tickit.core.management.event_router.Default_Wiring_Struct
A mapping of component output ports to component input ports with defaults
alias of
DefaultDict[ComponentID,DefaultDict[PortID,Set[tickit.core.typedefs.ComponentPort]]]
- tickit.core.management.event_router.Wiring_Struct
A mapping of component output ports to component input ports
alias of
Dict[ComponentID,Dict[PortID,Set[tickit.core.typedefs.ComponentPort]]]
- tickit.core.management.event_router.Default_InverseWiring_Struct
A mapping of component input ports to component output ports with defaults
alias of
DefaultDict[ComponentID,DefaultDict[PortID,tickit.core.typedefs.ComponentPort]]
- tickit.core.management.event_router.Inverse_Wiring_Struct
A mapping of component input ports to component output ports
alias of
Dict[ComponentID,Dict[PortID,tickit.core.typedefs.ComponentPort]]
- class tickit.core.management.event_router.Wiring(wiring: Optional[Dict[ComponentID, Dict[PortID, Set[tickit.core.typedefs.ComponentPort]]]] = None)[source]
A mapping of component output ports to component input ports with defaults.
A mapping of component output ports to component input ports, used to represent the connections between components in a system. Defaults are generated in case of missing mapping keys; for an unknown component, an empty output port mapping is created, whilst for an unknown output port an empty input port set is created.
A constructor which adds defaults to a wiring struct if provided.
A constructor which adds defaults to a wiring struct if provided, otherwise an empty default dict is created.
- Parameters
wiring (Optional[Wiring_Struct]) – An optional wiring struct. Defaults to None.
- Return type
- classmethod from_inverse_wiring(inverse_wiring: tickit.core.management.event_router.InverseWiring) tickit.core.management.event_router.Wiring[source]
Un-inverts an inverse wiring object to produce a Wiring instance.
- Parameters
inverse_wiring (InverseWiring) – A mapping of component input ports to component output ports.
- Returns
A mapping of component output ports to component input ports.
- Return type
- class tickit.core.management.event_router.InverseWiring(wiring: Optional[Dict[ComponentID, Dict[PortID, tickit.core.typedefs.ComponentPort]]] = None)[source]
A mapping of component input ports to component output ports.
A mapping of component input ports to component output ports, used to represent the connections between components in a system. When either an unknown input component or an unknown input component port is requested a KeyError is raised.
A constructor which adds defaults to an inverse wiring struct if provided.
A constructor which adds defaults to an inverse wiring struct if provided, otherwise an empty default dict is created.
- Parameters
wiring (Optional[Inverse_Wiring_Struct]) – An optional inverse wiring struct. Defaults to None.
- Return type
- classmethod from_wiring(wiring: tickit.core.management.event_router.Wiring) tickit.core.management.event_router.InverseWiring[source]
Inverts a Wiring object to create an InverseWiring instance.
- Parameters
wiring (Wiring) – A mapping of component output ports to component input ports.
- Returns
A mapping of component input ports to component output ports.
- Return type
- classmethod from_component_configs(configs: Iterable[tickit.core.components.component.ComponentConfig]) tickit.core.management.event_router.InverseWiring[source]
Creates an InverseWiring instance from an iterable of ComponentConfigs.
- Parameters
configs (Iterable[ComponentConfig]) – An iterable of component config data containers.
- Returns
A mapping of component input ports to component output ports.
- Return type
- class tickit.core.management.event_router.EventRouter(wiring: tickit.core.management.event_router.Wiring)[source]
- class tickit.core.management.event_router.EventRouter(wiring: tickit.core.management.event_router.InverseWiring)
A utility class responsible for routing changes between components.
An EventRouter constructor which stores possibly inverse wiring for use in utilities.
- Parameters
wiring (Union[Wiring, InverseWiring]) – A wiring or inverse wiring object representing the connections between components in the system.
- Return type
- property wiring: tickit.core.management.event_router.Wiring
The cached wiring used by the event router.
- Returns
The wiring used by the event router.
- Return type
- property components: Set[ComponentID]
A cached set of all components in the wiring.
- Returns
A set of all components in the wiring.
- Return type
Set[ComponentID]
- property output_components: Set[ComponentID]
A cached set of components which provide outputs.
- Returns
A set of components which provide outputs.
- Return type
Set[ComponentID]
- property input_components: Set[ComponentID]
A cached set of components which recieve inputs.
- Returns
A set of components which recieve inputs.
- Return type
Set[ComponentID]
- property component_tree: Dict[ComponentID, Set[ComponentID]]
A cached mapping of first order component dependants.
A cached property which returns a mapping of components to the set of components which are wired to any of its outputs.
- Returns
A mapping of components to the set of components which are wired to any of its outputs.
- Return type
Dict[ComponentID, Set[ComponentID]]
- property inverse_component_tree: Dict[ComponentID, Set[ComponentID]]
A cached mapping of first order component dependancies.
A cached property which returns a mapping of components to the set of components which are wired to any of its inputs.
- Returns
A mapping of components to the set of components which are wired to any of its inputs.
- Return type
Dict[ComponentID, Set[ComponentID]]
- dependants(root: ComponentID) Set[ComponentID][source]
Finds set of all components which are recursively dependant on the root component.
- Parameters
root (ComponentID) – The root component.
- Returns
A set of all components which are recursively dependant on the root component.
- Return type
Set[ComponentID]
- route(source: ComponentID, changes: Mapping[PortID, object]) DefaultDict[ComponentID, Dict[PortID, object]][source]
Generates a mapping of input changes resulting from output changes.
A method which generates a mapping of input changes which result from the propagation of output changes according to the wiring.
- Parameters
source (ComponentID) – The source component of the output changes.
changes (Mapping[PortID, Hashable]) – A mapping of changes to the outputs of the source component.
- Returns
A mapping of the input changes which result from the propagation of output changes according to the wiring.
- Return type
DefaultDict[ComponentID, Dict[PortID, Hashable]]
tickit.core.management.ticker
- class tickit.core.management.ticker.Ticker(wiring: Union[tickit.core.management.event_router.Wiring, tickit.core.management.event_router.InverseWiring], update_component: Callable[[tickit.core.typedefs.Input], Awaitable[None]])[source]
A utility class responsible for sequencing the update of components during a tick.
A utility class responsible for sequencing the update of components during a tick by eagerly updating each component which has had all of its dependencies resolved.
A Ticker constructor which creates an event router and performs initial setup.
- Parameters
wiring (Union[Wiring, InverseWiring]) – A wiring or inverse wiring object representing the connections between components in the system.
update_component (Callable[[Input], Awaitable[None]]) – A function or method which may be called to request a component performs and update, such updates should result in a subsequent call to the propagate method of the ticker.
- Return type
See also
- async schedule_possible_updates() None[source]
Updates components with resolved dependencies.
An asynchronous method which schedules updates for components with resolved dependencies, as determined by the intersection of the components first order dependencies and the set of componets which still require an update.
- Return type
- async propagate(output: tickit.core.typedefs.Output) None[source]
Propagates the output of an updated component.
An asynchronous message which propagates the output of an updated component by removing the component from the set of components requiring update, adding the routed inputs to the accumulator, and scheduling any possible updates. If no components require update the finsihed flag will be set.
- property components: Set[ComponentID]
The set of all components in the wiring.
- Returns
The set of all components in the wiring.
- Return type
Set[ComponentID]
tickit.core.state_interfaces
tickit.core.state_interfaces.internal
- tickit.core.state_interfaces.internal.C
A consumable value
alias of TypeVar(‘C’)
- tickit.core.state_interfaces.internal.P
A producable value
alias of TypeVar(‘P’)
- class tickit.core.state_interfaces.internal.Message(value: Any)[source]
An immutable data container for internal messages.
Create new instance of Message(value,)
- Parameters
value (Any) –
- property value
Alias for field number 0
- tickit.core.state_interfaces.internal.Messages
A list of messages
alias of
List[tickit.core.state_interfaces.internal.Message]
- class tickit.core.state_interfaces.internal.InternalStateServer(*args: Any, **kwargs: Any)[source]
A singleton, in memory, publish/subscribe message store and router.
A singleton, in memory, publish/subscribe message store and router. The single instance of this class holds a mapping of topics and a list of messages which have been pushed to them and a mapping of consumers which subscribe to the topics. Upon subscribing to a topic, a consumer is sent all of the messages which were previously registered against the topic. Upon pushing to a topic, the message is added to the store and immediately forwarded to each of the consumers which subscribe to the topic.
- Parameters
args (Any) –
kwargs (Any) –
- Return type
- async push(topic: str, message: tickit.core.state_interfaces.internal.Message) None[source]
An asynchronous method which propagates a message to subscribers and stores it.
An asynchronous method which propagates the given message to consumers which subscribe to the topic and stores the message in the topic -> messages mapping.
- async subscribe(consumer: tickit.core.state_interfaces.internal.InternalStateConsumer, topics: Iterable[str])[source]
Subscribes the consumer to the given topics, so it is notified a message is added.
An asynchronous method which adds a consumer to the subscriber list of each topic in the topics iterable. On subscription, previous messages on the topic are immediately passed to the consumer.
- Parameters
consumer (InternalStateConsumer) – The consumer which is subscribing to the topic.
topics (Iterable[str]) – The topic which the consumer is to be subscribed to.
- class tickit.core.state_interfaces.internal.InternalStateConsumer(callback: Callable[[tickit.core.state_interfaces.internal.C], Awaitable[None]])[source]
An internal, singleton based, implementation of the StateConsumer protocol.
A internal, singleton based, implementation of the StateConsumer protocol, this consumer can subscribe to InternalStateServer topics, upon recieving a message the consumer passes the value to the callback function passed during initialization, if a topic is subscribed to which does not yet exist it is created.
Gets an instance of the InternalStateServer for use in subscribe.
- Parameters
callback (Callable[[C], Awaitable[None]]) – An asynchronous handler function for consumed values.
- class tickit.core.state_interfaces.internal.InternalStateProducer[source]
An internal, singleton based, implementation of the StateProducer protocol.
An internal, singleton based, implementation of the StateProducer protocol, this producer can produce a value to a InternalStateServer topic, if the topic does not yet exist it is created.
Gets an instance of the InternalStateServer for use in produce.
tickit.core.state_interfaces.kafka
- tickit.core.state_interfaces.kafka.C
A consumable value
alias of TypeVar(‘C’)
- tickit.core.state_interfaces.kafka.P
A producable value
alias of TypeVar(‘P’)
- class tickit.core.state_interfaces.kafka.KafkaStateConsumer(callback: Callable[[tickit.core.state_interfaces.kafka.C], Awaitable[None]])[source]
A kafka implementation of the StateConsumer protocol.
A kafka implementation of the StateConsumer protocol, this consumer can subscribe to kafka topics, upon recieving a message the consumer passes the value to the callback function passed during initialization, if a topic is subscribed to which does not yet exist it is created.
Creates an AIOKafka Consumer and begins consuming subscribed topics.
- class tickit.core.state_interfaces.kafka.KafkaStateProducer[source]
A kafka implementation of the StateProducer protocol.
A kafka implementation of the StateProducer protocol, this producer can produce a value to a kafka topic, if the topic does not yet exist it is created.
Creates and starts and AIOKafka Producer.
tickit.core.state_interfaces.state_interface
- tickit.core.state_interfaces.state_interface.C
A consumable value
alias of TypeVar(‘C’, covariant=True)
- tickit.core.state_interfaces.state_interface.P
A producable value
alias of TypeVar(‘P’, contravariant=True)
- class tickit.core.state_interfaces.state_interface.StateConsumer(*args, **kwds)[source]
An interface for types which implent publish/subscribe message consumers.
An interface for types which implement publish/subscribe message consumers, the consumer must be able to subscribe to topics within the messaging framework, upon recieving a message the consumer should pass the value to the callback function, if a topic is subscribed to which does not yet exist it should be created.
- class tickit.core.state_interfaces.state_interface.StateProducer(*args, **kwds)[source]
An interface for types which implement publish/subscribe message producers.
An interface for types which implment publish/subscribe message producers, the producer must be able to produce a value to a topic within the messaging framework, if the topic does not yet exist it should be created.
- tickit.core.state_interfaces.state_interface.StateInterface
The union of StateConsumer and StateProducer
alias of TypeVar(‘StateInterface’, tickit.core.state_interfaces.state_interface.StateConsumer, tickit.core.state_interfaces.state_interface.StateProducer)
- tickit.core.state_interfaces.state_interface.add(name: str, external: bool) Callable[[Type[tickit.core.state_interfaces.state_interface.StateInterface]], Type[tickit.core.state_interfaces.state_interface.StateInterface]][source]
A decorator to add a StateInterface to the registry.
A decorator to add a StateInterface to the registry of StateConsumers or StateProducer according to it’s signature. StateConsumers and StateProducers which are intended to work together should be added with the same name.
- Parameters
- Return type
Callable[[Type[tickit.core.state_interfaces.state_interface.StateInterface]], Type[tickit.core.state_interfaces.state_interface.StateInterface]]
- tickit.core.state_interfaces.state_interface.interfaces(external: bool = False) Set[str][source]
Gets a set of interface names for which both a StateConsumer and StateProducer exist.
Gets a set of interface names for which both a StateConsumer and StateProducer exist. The external option may be used to restrict interfaces to those which may be used in simulations which are distriubted across multiple processes.
- tickit.core.state_interfaces.state_interface.satisfy_externality(external: bool, interfaces: Dict[str, Tuple[Type[tickit.core.state_interfaces.state_interface.StateInterface], bool]]) Set[str][source]
Finds which interfaces satisfy the externality requirement provided.
- Parameters
external (bool) – If true, only interfaces which can be used in simulations which are distributed across processes are returned. If false, all interfaces are returned.
interfaces (Dict[str, Tuple[Type[StateInterface], bool]]) – A mapping of interface names to a tuple of their class and their externality flag.
- Returns
A set of interface names which satisfy the externality requirement.
- Return type
Set[str]
- tickit.core.state_interfaces.state_interface.get_interface(name: str) Tuple[Type[tickit.core.state_interfaces.state_interface.StateConsumer], Type[tickit.core.state_interfaces.state_interface.StateProducer]][source]
Get the StateConsumer and StateProducer classes for a given interface name.
- Parameters
name (str) – The name of the interface to be retrieved.
- Returns
A tuple of the StateConsumer and StateProducer classes.
- Return type
Tuple[Type[StateConsumer], Type[StateProducer]]
tickit.core.device
- tickit.core.device.InMap
A bound typevar for mappings of device inputs
- tickit.core.device.OutMap
A bound typevar for mappings of device outputs
- class tickit.core.device.DeviceUpdate(outputs: tickit.core.device.OutMap, call_at: Optional[SimTime])[source]
An immutable data container for Device outputs and callback request time.
- Parameters
outputs – A mapping of device output keys and current values.
call_at – The simulation time at which the component requests to be awoken.
- class tickit.core.device.Device(*args, **kwds)[source]
An interface for types which implement simulated devices.
- update(time: SimTime, inputs: tickit.core.device.InMap) tickit.core.device.DeviceUpdate[tickit.core.device.OutMap][source]
A method which implements device behaviour according to the time and its inputs.
A method which implements the (typically physics based) changes which occur within the device in response to either the progression of time or the alteration of inputs to the device. The method returns the new observable state of the device and may optionally include a time in which the method should be called again.
- Parameters
time (SimTime) – The current simulation time (in nanoseconds).
inputs (tickit.core.device.InMap) – A mapping of device inputs and their values.
- Return type
- class tickit.core.device.DeviceConfig[source]
A data container for device configuration.
A data container for device configuration which acts as a named union of subclasses to facilitate automatic deserialization.
- Return type
- static configures() Type[tickit.core.device.Device][source]
A static method which returns the Device class configured by this config.
- Returns
The Device class configured by this config.
- Return type
Type[Device]
tickit.core.adapter
- tickit.core.adapter.T
Message type
alias of TypeVar(‘T’)
- class tickit.core.adapter.Adapter(*args, **kwds)[source]
An interface for types which implement device adapters.
- async run_forever() None[source]
An asynchronous method allowing indefinite running of core adapter logic.
An asynchronous method allowing for indefinite running of core adapter logic (typically the hosting of a protocol server and the interpretation of commands which are supplied via it).
- Return type
- class tickit.core.adapter.ListeningAdapter(*args, **kwds)[source]
An interface for adapters which require to be notified after a device updates.
- class tickit.core.adapter.AdapterConfig[source]
A data container for adapter configuration.
A data container for adapter configuration which acts as a named union of subclasses to facilitate automatic deserialization.
- Return type
- static configures() Type[tickit.core.adapter.Adapter][source]
A static method which returns the Adapter class configured by this config.
- Returns
The Adapter class configured by this config.
- Return type
Type[Adapter]
- class tickit.core.adapter.ConfigurableAdapter[source]
A mixin used to create an adapter with a configuration data container.
- class tickit.core.adapter.Interpreter(*args, **kwds)[source]
An interface for types which handle messages recieved by an adapter.
- async handle(adapter: tickit.core.adapter.Adapter, message: tickit.core.adapter.T) Tuple[AsyncIterable[tickit.core.adapter.T], bool][source]
An asynchronous method which handles messages recieved by an adapter.
An asynchronous method which handles messages recieved by an adapter, replies are sent as an asynchronous iterable to support setting of continious readback, stand alone replies should be wrapped in an asynchronous iterable of length one.
- class tickit.core.adapter.Server(*args, **kwds)[source]
An interface for types which implement an external messaging protocol.
- async run_forever(on_connect: Callable[[], AsyncIterable[Optional[tickit.core.adapter.T]]], handler: Callable[[tickit.core.adapter.T], Awaitable[AsyncIterable[Optional[tickit.core.adapter.T]]]]) None[source]
An asynchronous method allowing indefinite running of core server logic.
- Parameters
- Return type
- class tickit.core.adapter.ServerConfig[source]
A data container for server configuration.
A data container for server configuration which acts as a named union of subclasses to facilitate automatic deserialization.
- Return type
- static configures() Type[tickit.core.adapter.Server][source]
A static method which returns the Adapter class configured by this config.
- Returns
The Server class configured by this config.
- Return type
Type[Server]
tickit.core.lifetime_runnable
- class tickit.core.lifetime_runnable.LifetimeRunnable(*args, **kwds)[source]
An interface for types which implement an awaitable run_forever method.
- tickit.core.lifetime_runnable.run_all(runnables: Iterable[tickit.core.lifetime_runnable.LifetimeRunnable]) List[_asyncio.Task][source]
Asynchronously runs the run_forever method of each lifetime runnable.
Creates and runs an asyncio task for the run_forever method of each lifetime runnable. Calls to the run_forever method are wrapped with an error handler.
- Parameters
runnables (Iterable[tickit.core.lifetime_runnable.LifetimeRunnable]) – An iterable of objects which implement run_forever.
- Returns
A list of asyncio tasks for the runnables.
- Return type
List[asyncio.Task]
- async tickit.core.lifetime_runnable.run_all_forever(runnables: Iterable[tickit.core.lifetime_runnable.LifetimeRunnable]) None[source]
Asynchronously runs the run_forever method of each lifetime runnable.
Creates and runs an asyncio task for the run_forever method of each lifetime runnable. Calls to the run_forever method are wrapped with an error handler. This function blocks until all run_forever methods have completed.
- Parameters
runnables (Iterable[tickit.core.lifetime_runnable.LifetimeRunnable]) – An iterable of objects which implement run_forever.
- Return type
tickit.core.typedefs
- tickit.core.typedefs.PortID
An identifier which specifies the input/output port of a component
alias of
str
- tickit.core.typedefs.State
A mapping of PortID to component input/output property
- tickit.core.typedefs.Changes
A mapping of the difference between State mappings
alias of
immutables._map.Map
- class tickit.core.typedefs.ComponentPort(component: ComponentID, port: PortID)[source]
An immutable dataclass for custom (de)serialization of component - port pairs.
- Parameters
component (ComponentID) –
port (PortID) –
- Return type
- serialize() str[source]
An apischema serialization method which returns a string of component:port.
- Returns
The serialized ComponentPort, in format component:port.
- Return type
- static deserialize(data: str) tickit.core.typedefs.ComponentPort[source]
An apischema deserialization method which builds a from a string of component:port.
- Returns
The deserialized ComponentPort.
- Return type
- Parameters
data (str) –
- class tickit.core.typedefs.Input(target: ComponentID, time: SimTime, changes: Changes)[source]
An immutable data container for Component inputs.
- Parameters
target (ComponentID) – The Component which is to handle the Input.
time (SimTime) – The simulation time at which the Input was produced and is to be handled.
changes (Changes) – The changes to the component inputs.
- Return type
- class tickit.core.typedefs.Output(source: ComponentID, time: SimTime, changes: Changes, call_at: Optional[SimTime])[source]
An immutable data container for Component outputs.
- Parameters
source (ComponentID) – The Component which produced the Output.
time (SimTime) – The simulation time at which the Output was produced and is to be handled.
changes (Changes) – The changes to the component outputs.
call_in – The duration after which the component requests to be awoken.
call_at (Optional[SimTime]) –
- Return type
- class tickit.core.typedefs.Interrupt(source: ComponentID)[source]
An immutable data container for scheduling Component interrupts.
- Parameters
component – The Component which is requesting an interrupt.
source (ComponentID) –
- Return type
tickit.devices
tickit.devices.sink
- class tickit.devices.sink.Sink[source]
A simple device which can take any input and produces no output.
A constructor of the sink, with no arguments.
- Return type
- class Inputs(**kwargs)
A typed mapping containing the ‘input’ input value
- class Outputs(**kwargs)
An empty typed mapping of device outputs
- update(time: SimTime, inputs: tickit.devices.sink.Inputs) tickit.core.device.DeviceUpdate[tickit.devices.sink.Outputs][source]
The update method which logs the inputs at debug level and produces no outputs.
- Parameters
- Returns
The produced update event which never contains any changes, and never requests a callback.
- Return type
tickit.devices.source
- class tickit.devices.source.Source(value: Any)[source]
A simple device which produces a pre-configured value.
A constructor of the source, which takes the pre-configured output value.
- Parameters
value (Any) – A pre-configured output value.
- Return type
- class Inputs(**kwargs)
An empty typed mapping of device inputs
- class Outputs(**kwargs)
A typed mapping containing the ‘value’ output value
- update(time: SimTime, inputs: tickit.devices.source.Inputs) tickit.core.device.DeviceUpdate[tickit.devices.source.Outputs][source]
The update method which produces the pre-configured output value.
- Parameters
- Returns
The produced update event which contains the pre-configured value, and never requests a callback.
- Return type
tickit.utils
tickit.utils.byte_format
- class tickit.utils.byte_format.ByteFormat(format: bytes)[source]
An immutable dataclass for custom (de)serialization byte format strings.
- serialize() str[source]
An apischema serialiazation method which returns a utf-8 decoded string.
- Returns
A utf-8 decoded string of the format.
- Return type
- static deserialize(data: str) tickit.utils.byte_format.ByteFormat[source]
An apischema deserialization method builds from a utf-8 encoded string.
- Returns
The deserialized ByteFormat.
- Return type
- Parameters
data (str) –
tickit.utils.configuration
tickit.utils.configuration.configurable
- tickit.utils.configuration.configurable.Func
A function or method
alias of TypeVar(‘Func’, bound=
Callable)
- tickit.utils.configuration.configurable.T
A configurable type
alias of TypeVar(‘T’, covariant=True)
- tickit.utils.configuration.configurable.rec_subclasses(cls: type) Iterator[type][source]
Recursive implementation of type.__subclasses__.
- Parameters
cls (Type) – The base class.
- Returns
An iterator of subclasses.
- Return type
Iterator[type]
- tickit.utils.configuration.configurable.configurable_alias(sub: Type) str[source]
Gets the alias of a configurable sub-class.
- Parameters
sub (Type) – The sub-class to be aliased.
- Returns
The alias assigned to the sub-class.
- Return type
- tickit.utils.configuration.configurable.configurable_base(cls: tickit.utils.configuration.configurable.Cls) tickit.utils.configuration.configurable.Cls[source]
A decorator to make a config base class which can deserialize aliased sub-classes.
A decorator which makes a config class the root of a tagged union of sub-classes allowing for serialization and deserialization of config trees by class alias. The function registers both an apischema serialization and an apischema deserialization conversion for the base class which perform lookup based on a tagged union of aliased sub-classes.
- class tickit.utils.configuration.configurable.Config(*args, **kwds)[source]
An interface for types which implement configurations.
- static configures() Type[tickit.utils.configuration.configurable.T][source]
A static method which returns the class configured by this config.
- Returns
The class configured by this config.
- Return type
Type[T]
- tickit.utils.configuration.configurable.configurable(template: Type, ignore: Sequence[str] = []) Callable[[Type], Type][source]
A decorator to add a config data container sub-class to a class.
A decorator to make a class configurable by adding a config data container sub-class which is typically registered against a configurable base class allowing for serialization & deserialization by union tagging.
- Parameters
template (Type) – A template dataclass from which fields are borrwoed
ignore (Sequence[str]) – Fields which should not be serialized / deserialized. Defaults to [].
- Returns
A decorator which adds the config data container.
- Return type
Callable[[Type], Type]
tickit.utils.configuration.loading
- tickit.utils.configuration.loading.read_configs(config_path) List[tickit.core.components.component.ComponentConfig][source]
A utility function which reads and deserializes configs.
A utility function which reads config files, performs yaml deserialization, loads the required internal and external modules then subsequently performs apischema deserialization to produce a list of component configuration objects.
- Parameters
config_path ([type]) – The path to the config file.
- Returns
A list of component configuration objects.
- Return type
List[ComponentConfig]
tickit.utils.singleton
- class tickit.utils.singleton.Singleton[source]
A singlton metaclass, which allows only one instance of derived types.
- __call__(*args: Any, **kwargs: Any) tickit.utils.singleton.Singleton[source]
Return the class instance, if uninstantiated create an instance.
- Parameters
args (Any) –
kwargs (Any) –
- Return type
tickit.utils.topic_naming
- tickit.utils.topic_naming.valid_component_id(component: ComponentID) None[source]
Checks the validity of a ComponentID for generating a topic name.
- Parameters
component (ComponentID) – The ComponentID to check.
- Raises
ValueError – The component name is empty.
- Return type
- tickit.utils.topic_naming.output_topic(component: ComponentID) str[source]
Creates the output topic name for a given component.
- Parameters
component (ComponentID) – The component for which an output topic name is required.
- Returns
The output topic name of the component.
- Return type
- tickit.utils.topic_naming.input_topic(component: ComponentID) str[source]
Creates returns the input topic name for a given component.
- Parameters
component (ComponentID) – The component for which an input topic name is required.
- Returns
The input topic name of the component.
- Return type