BitstreamEvolutionProtocols.py

Core protocols and interfaces for the BitstreamEvolution framework.

Defines the abstract contracts (Protocol classes) that all concrete implementations must satisfy: Individual, Circuit, Population, Fitness, Hardware, Reproducer, GenerateMeasurements, and EvaluatePopulationFitness. Start here to understand the system architecture.

class BitstreamEvolutionProtocols.Circuit(*args, **kwargs)

The most basic Class Protocol for creating the Circuits that are evaluated on physical FPGAs. This may or may not be an Individual based on what the Individual represents. It will if the Individual represents a circuit in its entirety. It will not if you are simultaniously evolving multiple sub-sections that need to be combined to make the circuit to be evaluated.

compile(fpga: FPGA_Compilation_Data) → Result[None, Exception]

This looks at the fpga and compiles the circuit for it if it can. If it can it does all of its work in the working_dir and returns with Ok(Path) for the path to the file/directory containing this data. If it cannot, it should return an exception Err(Exception) explaining why it failed. This should never raise an Exception, only return one as described.

class BitstreamEvolutionProtocols.CircuitFactory(*args, **kwargs)

The most basic Protocol for turning Individuals into Circuits in whatever way best fits your application. How the circuit is built should be fully specified here, and any unique roles the individuals have should be specified here; however, how these individuals are selected and matched to roles is not.

__call__(populations: list[Population]) → dict[Circuit, list[tuple[Population, Individual]]]

This takes the population of Individuals and constructs the necessary Circuit from it as requested. It returns the circuits as keys in a dictionary, where the associated values are a list of tupples for each Individual used to generate the circuit (or all individuals whose fitness is impacted directly by the circuit’s fitness) and includes the population the individual was from, and the individual itself.

class BitstreamEvolutionProtocols.DataRequest(value)

class BitstreamEvolutionProtocols.EvaluatePopulationFitness(*args, **kwargs)

Fully Evaluates a Population, the fitnesses in the population are fully specified. The populations involved will be edited in place. Any populations not provided will not be edited.

class BitstreamEvolutionProtocols.FPGA_Compilation_Data(model: FPGA_Model, id: str)

This contains all data needed to compile data for a particular FPGA.

model: FPGA_Model

The id for the particular FPGA (Maybe?)

class BitstreamEvolutionProtocols.FPGA_Model(value)

All of the FPGA models that any part of our code supports

class BitstreamEvolutionProtocols.Fitness(*args, **kwargs)

The most basic Class Protocol that contains the result of an evaluation, allowing you to compare the resulting fitness. This should match most numeric types (i.e. int, float) but allows you to do something more complex as desired.

The slash here ensures that the arguments before it can only be provided as positional arguments.

class BitstreamEvolutionProtocols.GenData(generation_number: int)

The most basic Generation Data Object. This simply requires a generation_number.

class BitstreamEvolutionProtocols.GenDataFactory(*args, **kwargs)

The most basic Function Protocol that converts the Generation Data from the previous iteration to the one for the next generation. It also constructs the initial GenData object (gen_data is None) and determines when the final generation occours (returns None).

class BitstreamEvolutionProtocols.GenerateInitialPopulation(*args, **kwargs)

Somehow gets you an initial implementation.

class BitstreamEvolutionProtocols.GenerateMeasurements(*args, **kwargs)

Generate the measurements to take for the given populations. Returns a dict where all new measurements are given, and map to the individuals whose fitnesses they impact and the population the individuals are in.

class BitstreamEvolutionProtocols.Hardware(*args, **kwargs)

Used to Evaluate Measurements. Compile hardware would be responsible for compiling the Circuit in the Measurement object passed to it in request_measurement().

Intended concurrency model (server-client):

request_measurement() is async so that multiple measurements can be dispatched concurrently via asyncio.gather() in Evolution.run(). This is meaningful when Hardware is a client that sends requests over a network to a hardware server — the await genuinely suspends while waiting for the network response, allowing other coroutines to run in the meantime.

The current Microcontroller implementation uses blocking pyserial and does NOT achieve real concurrency. As the codebase moves to a server-client model, asyncio.gather() will provide true parallelism across multiple FPGAs automatically. The server-side should use serial_asyncio (https://pypi.org/project/serial-asyncio/) in place of pyserial so that serial reads are genuinely non-blocking within the server’s event loop.

Per-FPGA exclusivity constraint:

Each physical Icestick (or FPGA device) can only be accessed by one process at a time — iceprog holds exclusive USB access while programming the device. Implementations of this protocol MUST ensure that concurrent calls to request_measurement() targeting the same physical FPGA are serialized. Recommended approaches for the server:

Option A — asyncio.Semaphore(1) per FPGA:

Each FPGA gets its own Semaphore. request_measurement acquires the semaphore for the target device before proceeding. Requests for different FPGAs run concurrently; requests for the same FPGA queue up automatically.

Option B — Per-FPGA asyncio.Queue with a dedicated worker coroutine:

One worker coroutine per FPGA dequeues and processes measurements one at a time. Naturally serializes access while allowing inter-FPGA parallelism.

See .claude/docs/hardware_concurrency.md for full discussion.

class BitstreamEvolutionProtocols.Individual(*args, **kwargs)

The most basic Class Protocol for creating the Individuals that evolution will act upon. This specifies basically nothing.

class BitstreamEvolutionProtocols.Measurement(FPGA_request: str, data_request: DataRequest, circuit_to_measure: Circuit, num_samples: int)

All measurement data, this could even be a class potentially

__init__(FPGA_request: str, data_request: DataRequest, circuit_to_measure: Circuit, num_samples: int) → None

Todo

Figure out the format for an FPGA Request, potentially also changing the type, and adjust that here.

exception BitstreamEvolutionProtocols.MeasurementError

exception BitstreamEvolutionProtocols.MeasurementNotTaken

class BitstreamEvolutionProtocols.Population(individuals: Iterable[Individual], fitnesses: Iterable[Fitness | None] | None = None)

This is the Population object used to hold individuals and their fitnesses durring evolution. It starts out with its full list of individuals, and optionally fitnesses. No individuals in the list may be duplicates. (determined using == ) If there is no fitness it is None. Fitnesses can be added to individuals in the population as desired, and once all individuals are added, the population can be sorted. The Population can also be itterated through. (iter() -then-> next())

__init__(individuals: Iterable[Individual], fitnesses: Iterable[Fitness | None] | None = None)

Can raise ValueError if Individuals are not unique.

set_fitness(individual: Individual, fitness: Fitness) → None

This can return value error if provided individual is not in the population.

sort(key: Callable[[Fitness], Fitness], reverse: bool) → None

Operates on the Population fitness function like the standard sort for a list. May Raise TypeError if population not fully evaluated.

class BitstreamEvolutionProtocols.Reproducer(*args, **kwargs)

Gets a population and returns another population filled with the children of this generation. (reproduce + mutation)