Source code for labgrid.remote.scheduler

from collections import defaultdict

import attr

[docs] @attr.s(eq=False) class TagSet: name = attr.ib(validator=attr.validators.instance_of(str)) tags = attr.ib(validator=attr.validators.instance_of(set))
[docs] def schedule_step(places, filters): "Find the filters that can be directly allocated without overlap." interest = defaultdict(list) for f in filters: for place in places: if f.tags.issubset(place.tags): interest[place].append(f) if not interest: return {} limit = min(map(len, interest.values())) allocation = {} for interest_place, interest_filters in interest.items(): if len(interest_filters) == limit: allocation[interest_filters.pop(0)] = interest_place return allocation
[docs] def schedule_overlaps(places, filters): "Iterate schedule_step until no more allocations are found." places = places[:] filters = filters[:] allocation = {} while True: new = schedule_step(places, filters) if not new: break for f, place in new.items(): assert f not in allocation places.remove(place) filters.remove(f) allocation[f] = place return allocation
[docs] def schedule(places, filters): allocation = schedule_overlaps(places, filters) return { for f, p in allocation.items()}