slim.types.queue module

Cages, Farms and Organisations communicate via a mix of traditional API and message passing via channels. Here are some types and tools used to handle such queues.

class slim.types.queue.CageEvent

Bases: abc.ABC, slim.types.queue.Event

abstract property event_time

class slim.types.queue.DamAvailabilityBatch(availability_date: 'dt.datetime', geno_distrib: 'GenoDistrib')

Bases: slim.types.queue.CageEvent

availability_date: dt.datetime

property event_time

geno_distrib: GenoDistrib

class slim.types.queue.EggBatch(hatch_date: 'dt.datetime', geno_distrib: 'GenoDistrib')

Bases: slim.types.queue.CageEvent

property event_time

geno_distrib: GenoDistrib

hatch_date: dt.datetime

class slim.types.queue.Event

Bases: object

An empty parent class for all the events with a default serialiser.

to_json_dict()

class slim.types.queue.PriorityQueue

Bases: Generic[slim.types.queue.T]

Mutex-free priority queue. Only use where thread-safety is not required!

empty()

get()

put(item)

qsize()

class slim.types.queue.SamplingEvent(sampling_date: datetime.datetime)

Bases: slim.types.queue.Event

Internal sampling event used inside farm

sampling_date: datetime.datetime

class slim.types.queue.StepResponse(total_offspring: 'GenoDistrib', profit: 'float', total_cost: 'float', observation_space: 'ObservationSpace', loggable: 'dict')

Bases: object

loggable: dict

observation_space: ObservationSpace

profit: float

total_cost: float

total_offspring: GenoDistrib

class slim.types.queue.TravellingEggBatch(arrival_date: 'dt.datetime', hatch_date: 'dt.datetime', geno_distrib: 'GenoDistrib')

Bases: slim.types.queue.CageEvent

arrival_date: dt.datetime

property event_time

geno_distrib: GenoDistrib

hatch_date: dt.datetime

class slim.types.queue.TreatmentEvent(affecting_date: 'dt.datetime', treatment_type: 'Treatment', effectiveness_duration_days: 'int', first_application_date: 'dt.datetime', end_application_date: 'dt.datetime')

Bases: slim.types.queue.CageEvent

affecting_date: dt.datetime

effectiveness_duration_days: int

end_application_date: dt.datetime

property event_time

first_application_date: dt.datetime

treatment_type: Treatment

property treatment_window

slim.types.queue.pop_from_queue(queue: PriorityQueue[slim.types.queue.EventT], cur_time: datetime.datetime, continuation: Callable[[slim.types.queue.EventT], None])

Pops an event from a queue and call a continuation function

Parameters

• queue – the queue to process

• cur_time – the current time to compare the events with. Only events past cur_time will be popped

• continuation – the function to call for each event in the queue