API#
tickit#
This is the internal API reference for tickit
- tickit.__version__: str#
Version number as calculated by pypa/setuptools_scm
tickit.core#
tickit.core.adapter#
- tickit.core.adapter.T#
Message type
alias of TypeVar(‘T’)
- class tickit.core.adapter.RaiseInterrupt(*args, **kwargs)[source]#
A raise_interrupt function that should be passed to
Adapter.
- class tickit.core.adapter.Adapter[source]#
An interface for types which implement device adapters.
- async run_forever(device: Device, raise_interrupt: RaiseInterrupt) 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).
- class tickit.core.adapter.Interpreter[source]#
An interface for types which handle messages recieved by an adapter.
- abstract async handle(adapter: Adapter, message: T) Tuple[AsyncIterable[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[source]#
An interface for types which implement an external messaging protocol.
- async run_forever(on_connect: Callable[[], AsyncIterable[Optional[T]]], handler: Callable[[T], Awaitable[AsyncIterable[Optional[T]]]]) None[source]#
An asynchronous method allowing indefinite running of core server logic.
- Parameters:
on_connect (Callable[[], AsyncIterable[Optional[T]]]) – An asynchronous iterable of messages to be sent once a client connects.
handler (Callable[[T], Awaitable[AsyncIterable[Optional[T]]]]) – An asynchronous method used to handle recieved messages, returning an asynchronous iterable of replies.
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: 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[source]#
An interface for types which implement simulated devices.
- abstract update(time: SimTime, inputs: InMap) DeviceUpdate[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 – The current simulation time (in nanoseconds).
inputs – A mapping of device inputs and their values.
tickit.core.runner#
- tickit.core.runner.run_all(awaitables: Iterable[Awaitable[None]]) List[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:
awaitables – An iterable of awaitable objects.
- Returns:
A list of asyncio tasks for the awaitables.
- Return type:
List[asyncio.Task]
- async tickit.core.runner.run_all_forever(awaitables: Iterable[Awaitable[None]]) 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:
awaitables – An iterable of awaitable objects.
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
Map
- class tickit.core.typedefs.ComponentPort(component: ComponentID, port: PortID)[source]#
An immutable dataclass for custom (de)serialization of component - port pairs.
- 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) ComponentPort[source]#
An apischema deserialization method which builds a from a string of component:port.
- Returns:
The deserialized ComponentPort.
- Return type:
- class tickit.core.typedefs.Input(target: ComponentID, time: SimTime, changes: Changes)[source]#
An immutable data container for Component inputs.
- Parameters:
target – The Component which is to handle the Input.
time – The simulation time at which the Input was produced and is to be handled.
changes – The changes to the component inputs.
- 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 – The Component which produced the Output.
time – The simulation time at which the Output was produced and is to be handled.
changes – The changes to the component outputs.
call_in – The duration after which the component requests to be awoken.
tickit.core.components#
tickit.core.components.component#
- class tickit.core.components.component.Component(name: ComponentID)[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).
- abstract async run_forever(state_consumer: Type[StateConsumer], state_producer: Type[StateProducer]) None[source]#
An asynchronous method allowing indefinite running of core logic.
- class tickit.core.components.component.ComponentConfig(name: ComponentID, inputs: Dict[PortID, 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.
- class tickit.core.components.component.BaseComponent(name: ComponentID)[source]#
A base class for compnents, implementing state interface related methods.
- async handle_input(input: 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.
- Parameters:
time (SimTime) – The current time (in nanoseconds).
changes (Changes) – A mapping of the difference between the last observable state and the current observable state.
call_at (Optional[SimTime]) – The simulation time at which the component requests to be awoken.
- 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.
- async run_forever(state_consumer: Type[StateConsumer], state_producer: Type[StateProducer]) None[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.
- Parameters:
time (SimTime) – The current simulation time (in nanoseconds).
changes (Changes) – A mapping of changed component inputs and their new values.
tickit.core.components.device_simulation#
- class tickit.core.components.device_simulation.DeviceSimulation(name: ~tickit.core.typedefs.ComponentID, device: ~tickit.core.device.Device, adapters: ~typing.List[~tickit.core.adapter.Adapter] = <factory>)[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.
- async run_forever(state_consumer: Type[StateConsumer], state_producer: Type[StateProducer]) None[source]#
Sets up state interfaces, runs adapters and blocks until any complete.
- 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 Adapters of the update, computes changes to the state of the component and sends the resulting Output.
- Parameters:
time (SimTime) – The current simulation time (in nanoseconds).
changes (Changes) – A mapping of changed component inputs and their new values.
tickit.core.components.system_simulation#
- class tickit.core.components.system_simulation.SystemSimulationComponent(name: ComponentID, components: List[ComponentConfig], expose: Dict[PortID, 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.
- components: List[ComponentConfig]#
A list of immutable component configuration data containers, used to construct internal components.
- expose: Dict[PortID, ComponentPort]#
A mapping of outputs which the system simulation exposes and the corresponding output of an internal component.
- async run_forever(state_consumer: Type[StateConsumer], state_producer: Type[StateProducer]) None[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.
- Parameters:
time (SimTime) – The current simulation time (in nanoseconds).
changes (Changes) – A mapping of changed component inputs and their new values.
- class tickit.core.components.system_simulation.SystemSimulation(name: ComponentID, inputs: Dict[PortID, ComponentPort], components: List[ComponentConfig], expose: Dict[PortID, ComponentPort])[source]#
Simulation of a nested set of components.
tickit.core.management#
tickit.core.management.schedulers#
tickit.core.management.schedulers.base#
- class tickit.core.management.schedulers.base.BaseScheduler(wiring: Union[Wiring, InverseWiring], state_consumer: Type[StateConsumer], state_producer: Type[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.
- 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.
- async update_component(input: Input) None[source]#
Sends a message with an input to the input topic of a component.
- Parameters:
input (Input) – The input to be included in the message sent to the component.
- async handle_message(message: Union[Interrupt, 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.
- 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.
- 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[Wiring, InverseWiring], state_consumer: Type[StateConsumer], state_producer: Type[StateProducer], initial_time: int = 0, simulation_speed: float = 1.0)[source]#
A master scheduler which orchestrates the running of a tickit simulation.
A master scheduler 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.
- async setup() None[source]#
Performs base setup and creates an awaitable flag to indicate new wakeups.
- 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.
- 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.
- 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.
tickit.core.management.schedulers.slave#
- class tickit.core.management.schedulers.slave.SlaveScheduler(wiring: Union[Wiring, InverseWiring], state_consumer: Type[StateConsumer], state_producer: Type[StateProducer], expose: Dict[PortID, 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.
- static add_exposing_wiring(wiring: Union[Wiring, InverseWiring], expose: Dict[PortID, ComponentPort]) 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: 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.
- Parameters:
input (Input) – The input message to be sent to the component.
- 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:
time (SimTime) – The current simulation time (in nanoseconds).
changes (Changes) – A mapping of changed component inputs and their new values.
- 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:
Tuple[Changes, Optional[SimTime]]
- async run_forever() None[source]#
Delegates to setup which instantiates the ticker and state interfaces.
- 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.
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[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[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,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,ComponentPort]]
- class tickit.core.management.event_router.Wiring(wiring: Optional[Dict[ComponentID, Dict[PortID, Set[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.
- classmethod from_inverse_wiring(inverse_wiring: InverseWiring) 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, 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.
- classmethod from_wiring(wiring: Wiring) 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[ComponentConfig]) 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: Wiring)[source]#
- class tickit.core.management.event_router.EventRouter(wiring: InverseWiring)
A utility class responsible for routing changes between components.
Construct an EventRouter.
An EventRouter stores possible 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.
- property wiring: 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 the 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[Wiring, InverseWiring], update_component: Callable[[Input], Coroutine[Any, Any, 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.
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.
- async propagate(output: 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.
- Parameters:
output (Output) – The output produced by the update of a component.
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,)
- 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.
- async push(topic: str, message: Message) None[source]#
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: 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.
- create_topic(topic: str) None[source]#
Creates a new topic as an empty message list.
- Parameters:
topic (str) – The topic to be created.
- class tickit.core.state_interfaces.internal.InternalStateConsumer(callback: Callable[[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[[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.
- Parameters:
callback (Callable[[C], Awaitable[None]]) – An asynchronous handler function for consumed values.
- 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, **kwargs)[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, **kwargs)[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[StateInterface]], Type[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.
- tickit.core.state_interfaces.state_interface.interfaces(external: bool = False) Set[str][source]#
Returns 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[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[StateConsumer], Type[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.adapters#
tickit.adapters.interpreters#
tickit.adapters.interpreters.command#
tickit.adapters.interpreters.command.command_interpreter#
- class tickit.adapters.interpreters.command.command_interpreter.Command(*args, **kwargs)[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: Adapter, message: AnyStr) Tuple[AsyncIterable, 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: 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”.
- async run_forever(on_connect: Callable[[], AsyncIterable[Optional[bytes]]], handler: Callable[[bytes], Awaitable[AsyncIterable[Optional[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.
tickit.adapters.composed#
- tickit.adapters.composed.T#
Message type
alias of TypeVar(‘T’)
- class tickit.adapters.composed.ComposedAdapter(server: Server, interpreter: 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.
- async on_connect() AsyncIterable[Optional[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: T) AsyncIterable[Optional[T]][source]#
Delegates message handling to the interpreter and raises interrupt if requested.
- Parameters:
message (T) – The message from the server to be handled.
- Returns:
An asynchronous iterable of reply messages.
- Return type:
AsyncIterable[Optional[T]]
- async run_forever(device: Device, raise_interrupt: RaiseInterrupt) None[source]#
Runs the server continously.
tickit.adapters.epicsadapter#
- class tickit.adapters.epicsadapter.EpicsAdapter(db_file: str, ioc_name: str)[source]#
An adapter implementation which acts as an EPICS IOC.
An EpicsAdapter constructor which stores the db_file path and the IOC name.
- link_input_on_interrupt(record: InputRecord, getter: Callable[[], Any]) None[source]#
Adds a record and a getter to the mapping of interrupting records.
- Parameters:
record (InputRecord) – The record to be added.
getter (Callable[[], Any]) – The getter handle.
- abstract on_db_load() None[source]#
Customises records that have been loaded in to suit the simulation.
- async run_forever(device: Device, raise_interrupt: RaiseInterrupt) None[source]#
Runs the server continously.
tickit.devices#
tickit.devices.sink#
- class tickit.devices.sink.SinkDevice[source]#
A simple device which can take any input and produces no output.
- class Inputs#
A typed mapping containing the ‘input’ input value
- class Outputs#
An empty typed mapping of device outputs
- update(time: SimTime, inputs: Inputs) DeviceUpdate[Outputs][source]#
The update method which logs the inputs at debug level and produces no outputs.
- Parameters:
time (SimTime) – The current simulation time (in nanoseconds).
inputs (State) – A mapping of inputs to the device and their values.
- Returns:
The produced update event which never contains any changes, and never requests a callback.
- Return type:
- class tickit.devices.sink.Sink(name: ComponentID, inputs: Dict[PortID, ComponentPort])[source]#
Arbitrary value sink that logs the value.
tickit.devices.source#
- class tickit.devices.source.SourceDevice(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.
- class Inputs#
An empty typed mapping of device inputs
- class Outputs#
A typed mapping containing the ‘value’ output value
- update(time: SimTime, inputs: Inputs) DeviceUpdate[Outputs][source]#
The update method which produces the pre-configured output value.
- Parameters:
time (SimTime) – The current simulation time (in nanoseconds).
inputs (State) – A mapping of inputs to the device and their values.
- Returns:
The produced update event which contains the pre-configured value, and never requests a callback.
- Return type:
- class tickit.devices.source.Source(name: ComponentID, inputs: Dict[PortID, ComponentPort], value: Any)[source]#
Source of a fixed value.
tickit.devices.cryostream#
tickit.devices.cryostream.base#
- class tickit.devices.cryostream.base.CryostreamBase[source]#
A base class for cryostream device logic.
A CryostreamBase constructor which assigns initial values.
- async ramp(ramp_rate: int, target_temp: int) None[source]#
Change gas temperature to a set value at a controlled rate.
- async plat(duration: int) None[source]#
Maintain the current temperature for a set amount of time.
- Parameters:
duration (int) – The duration for which the temperature should be held.
- async cool(target_temp: int) None[source]#
Make gas temperature decrease to a set value as quickly as possible.
- Parameters:
target_temp (int) – The target temperature.
- async end(ramp_rate: int) None[source]#
Bring the gas temperature to 300 K at ramp rate, then halt and stop.
- Parameters:
ramp_rate (int) – The rate at which the temperature should change.
- async turbo(turbo_on: int) None[source]#
Set turbo mode to use maximum achievable flow.
The system will use 10 l/min gas flow except above 310 K Above 310 K available heater power limits the maximum achievable flow to 5 l/min.
- Parameters:
turbo_on (int) – The desired turbo mode, where 0 denotes off and 1 denotes on.
- update_temperature(time: float) int[source]#
Update the Cryostream gas temperature according to mode and time.
- async set_status_format(status_format: int) None[source]#
Sets the status packet format.
- Parameters:
status_format (int) – The status packet format, where 0 denotes a standard status packet and 1 denotes an extended status packet.
- async get_status(status_format: int) Union[Status, ExtendedStatus][source]#
Get a Status or ExtendedStatus packet.
- Parameters:
status_format (int) – The status packet format, where 0 denotes a standard status packet and 1 denotes an extended status packet.
- Returns:
- Union[
tickit.devices.cryostream.status.Status, tickit.devices.cryostream.status.ExtendedStatus]: The status packet.
tickit.devices.cryostream.cryostream#
- class tickit.devices.cryostream.cryostream.CryostreamDevice[source]#
A Cryostream device, used for cooling of samples using cold gas.
A Cryostream constructor sets up initial internal values.
- class Inputs#
An empty typed mapping of device inputs
- class Outputs#
A typed mapping containing the ‘temperature’ output value
- update(time: SimTime, inputs: Inputs) DeviceUpdate[Outputs][source]#
The update method which changes the output temperature according to set modes.
- Returns:
The produced update event which contains the value of the output temperature, and requests callback if temperature should continue to change.
- Return type:
- class tickit.devices.cryostream.cryostream.CryostreamAdapter(host: str = 'localhost', port: int = 25565)[source]#
A Cryostream TCP adapter which sends regular status packets and can set modes.
A CryostreamAdapter constructor which instantiates a TcpServer with 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.
- async on_connect() AsyncIterable[bytes][source]#
A method which continiously yields status packets.
- Returns:
- An asyncronous iterable of packed Cryostream status
packets.
- Return type:
AsyncIterable[bytes]
- async turbo(turbo_on: bytes) None[source]#
A regex bytes command which enables / disables turbo mode.
- Parameters:
turbo_on (bytes) – The desired turbo mode, where 0 denotes off and 1 denotes on.
- async set_status_format(status_format: bytes) None[source]#
A regex bytes command which sets the status packet format.
- Parameters:
status_format (bytes) – The status packet format, where 0 denotes a standard status packet and 1 denotes an extended status packet.
- async plat(duration: bytes) None[source]#
A regex bytes command which maintains temperature for a set amount of time.
- Parameters:
duration (bytes) – The duration for which the temperature should be held.
- async end(ramp_rate: bytes) None[source]#
A regex bytes command which brings the gas temperature to 300 K at ramp rate.
- Parameters:
ramp_rate (bytes) – The rate at which the temperature should change.
tickit.devices.cryostream.status#
- class tickit.devices.cryostream.status.Status(length: int, type_status: int, gas_set_point: int, gas_temp: int, gas_error: int, run_mode: int, phase_id: int, ramp_rate: int, target_temp: int, evap_temp: int, suct_temp: int, remaining: int, gas_flow: int, gas_heat: int, evap_heat: int, suct_heat: int, line_pressure: int, alarm_code: int, run_time: int, controller_number: int, software_version: int, evap_adjust: int, status_bytes_string: str = '>BBHHhBBHHHHHBBBBBBHHBB')[source]#
A regular Cryostream status packet.
- class tickit.devices.cryostream.status.ExtendedStatus(length: int, type_status: int, gas_set_point: int, gas_temp: int, gas_error: int, run_mode: int, phase_id: int, ramp_rate: int, target_temp: int, evap_temp: int, suct_temp: int, remaining: int, gas_flow: int, gas_heat: int, evap_heat: int, suct_heat: int, line_pressure: int, alarm_code: int, run_time: int, controller_number: int, software_version: int, evap_adjust: int, turbo_mode: int, hardware_type: int, shutter_state: int, shutter_time: int, avg_gas_heat: int, avg_suct_heat: int, time_to_fill: int, total_hours: int, extended_packet_string: str = '>BBHHhBBHHHHHBBBBBBHHBBBBBBBBHH')[source]#
An extended Cryostream status packet.
- classmethod from_packed(b: bytes) ExtendedStatus[source]#
Create a status packet from its packed byte format.
tickit.device.cryostream.states#
- class tickit.devices.cryostream.states.RunModes(value)[source]#
An enumerator for Cryostream run modes.
- class tickit.devices.cryostream.states.PhaseIds(value)[source]#
An enumerator for Cryostream phases.
tickit.devices.femto#
tickit.devices.femto.femto#
- class tickit.devices.femto.femto.FemtoDevice(initial_gain: float, initial_current: float)[source]#
Electronic signal amplifier.
Initialise the Femto device class.
- Parameters:
- class Inputs#
An empty typed mapping of device inputs
- class Outputs#
A typed mapping containing the current output value
- set_gain(gain: float) None[source]#
Sets a new amplified difference between the input and output signals.
- Parameters:
gain (float) – The amplified difference between the input and output signals.
- get_gain() float[source]#
Returns the current amplified difference between the input and output signals.
- Returns:
the amplified difference between the input and output signals.
- Return type:
- set_current(input_current: float) None[source]#
Set the output current based on the input current and the gain.
- The output current is calculated as:
output_current = input_current * gain
- Parameters:
input_current (float) – The current of the input signal.
- get_current() float[source]#
Returns the output current of the signal.
- Returns:
The output current of the signal.
- Return type:
- update(time: SimTime, inputs: Inputs) DeviceUpdate[Outputs][source]#
Updates the state of the Femto device.
- Parameters:
time (SimTime) – The time of the simulation in nanoseconds.
inputs (dict) – The dict containing the input values of gain and current.
- Returns:
A container for the Device’s outputs and a callback time.
- Return type:
- class tickit.devices.femto.femto.FemtoAdapter(db_file: str, ioc_name: str)[source]#
The adapter for the Femto device.
An EpicsAdapter constructor which stores the db_file path and the IOC name.
tickit.devices.pneumatic#
tickit.devices.pneumatic.pneumatic#
- class tickit.devices.pneumatic.pneumatic.PneumaticDevice(initial_speed: float, initial_state: bool)[source]#
Pneumatic Device with movement controls.
Initialise a Pneumatic object.
- class Outputs#
A typed mapping containing the current output value
- set_speed(speed: float) None[source]#
Set the speed of movement for the device.
- Parameters:
speed (float) – The desired movement speed of the device.
- update(time: SimTime, inputs) DeviceUpdate[Outputs][source]#
Run the update logic for the device.
If the device is moving then the state of the device is updated. Otherwise nothing changes. In either case the current state of the device is returned.
- Parameters:
time (SimTime) – The time of the simulation in nanoseconds.
inputs (dict) – The dict containing the input values of state of the device.
- Returns:
A container for the Device’s outputs and a callback time.
- Return type:
- class tickit.devices.pneumatic.pneumatic.PneumaticAdapter(db_file: str, ioc_name: str)[source]#
An adapter for the Pneumatic class.
Connects the device to an external messaging protocol.
An EpicsAdapter constructor which stores the db_file path and the IOC name.
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) ByteFormat[source]#
An apischema deserialization method builds from a utf-8 encoded string.
- Returns:
The deserialized ByteFormat.
- Return type:
tickit.utils.configuration#
tickit.utils.configuration.configurable#
- 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.is_tagged_union: Dict[Type[Any], bool] = defaultdict(<function <lambda>>, {<class 'tickit.core.device.Device'>: True, <class 'tickit.core.adapter.Adapter'>: True, <class 'tickit.core.adapter.Interpreter'>: True, <class 'tickit.core.adapter.Server'>: True, <class 'tickit.core.components.component.Component'>: True, <class 'tickit.core.components.component.ComponentConfig'>: True})#
Whether the current class is registered as a tagged union
- tickit.utils.configuration.configurable.as_tagged_union(cls: Cls) 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.
- Parameters:
cls (Cls) – The config base class.
- Returns:
The modified config base class.
- Return type:
Cls
tickit.utils.configuration.loading#
- tickit.utils.configuration.loading.importing_conversion(typ: Type) Optional[Union[Conversion, Callable[[Any], Any], property, LazyConversion, Tuple[Union[Conversion, Callable[[Any], Any], property, LazyConversion], ...]]][source]#
Create a conversion that imports the module of a ComponentConfig.
When a ComponentConfig is requested from a dict, take its fully qualified name from the tagged union dict and import it before deserializing it
- tickit.utils.configuration.loading.read_configs(config_path) List[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.
tickit.utils.singleton#
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.