States

BaseState

Abstract base class for states.

Parameters:

Name	Type	Description	Default
name	str	Name of the state	required
is_actionable	bool	If True, state can be changed by a policy	False
is_observable	bool	If True, state can be observed by a policy	True
exception_on_nan	bool	If True an exception is raised whenever an instance tries to access the value of the state and this value is None.	False

Source code in lineflow/simulation/states.py

class BaseState(metaclass=ABCMeta):
    """
    Abstract base class for states.

    Args:
        name (str): Name of the state
        is_actionable (bool): If `True`, state can be changed by a policy
        is_observable (bool): If `True`, state can be observed by a policy
        exception_on_nan (bool): If `True` an exception is raised whenever an instance tries to
            access the value of the state and this value is `None`.
    """

    def __init__(self, name, is_actionable=False, is_observable=True, exception_on_nan=False):
        self.name = name
        self.is_actionable = is_actionable
        self.is_observable = is_observable
        self._value = None
        self.exception_on_nan = exception_on_nan

    @property
    def value(self):
        """
        The (scalar) value of the state.
        """
        if self._value is None and self.exception_on_nan:
            raise ValueError('NAN value detected')
        return self._value

    def _change_value(self, value):
        """
        Method that has to be called whenever the value of the state should be changed.
        """
        self.assert_valid(value)
        self._value = value

    def apply(self, value):
        """
        Should be called from a policy to change the value of the state
        """
        assert self.is_actionable, 'Trying to set a non-action'
        self._change_value(value)

    def update(self, value):
        """
        Should be called from a `lineflow.simulation.base.LineObject` to change the value
        of the state
        """
        self._change_value(value)

    def to_str(self):
        return str(self.value)

    def print(self):
        return self.to_str()

    def reverse(self, series):
        """
        To be implemented for states that hold categorical values
        """
        return series

    def assert_valid(self, value):
        """
        Can be implemented by a downstream class to check whether value to be set is valid
        """
        pass

value property

The (scalar) value of the state.

apply(value)

Should be called from a policy to change the value of the state

Source code in lineflow/simulation/states.py

def apply(self, value):
    """
    Should be called from a policy to change the value of the state
    """
    assert self.is_actionable, 'Trying to set a non-action'
    self._change_value(value)

assert_valid(value)

Can be implemented by a downstream class to check whether value to be set is valid

Source code in lineflow/simulation/states.py

def assert_valid(self, value):
    """
    Can be implemented by a downstream class to check whether value to be set is valid
    """
    pass

reverse(series)

To be implemented for states that hold categorical values

Source code in lineflow/simulation/states.py

def reverse(self, series):
    """
    To be implemented for states that hold categorical values
    """
    return series

update(value)

Should be called from a lineflow.simulation.base.LineObject to change the value of the state

Source code in lineflow/simulation/states.py

def update(self, value):
    """
    Should be called from a `lineflow.simulation.base.LineObject` to change the value
    of the state
    """
    self._change_value(value)

CountState

Bases: NumericState

State to count discrete events.

Parameters:

Name	Type	Description	Default
name	str	Name of the state	required
vmin	float	The allowed minimal value the state accepts	0
vmax	float	The allowed maximal value the state accepts	inf
is_actionable	bool	If True, state can be changed by a policy	False
is_observable	bool	If True, state can be observed by a policy	True
exception_on_nan	bool	If True an exception is raised whenever an instance tries to access the value of the state and this value is None.	False

Source code in lineflow/simulation/states.py

class CountState(NumericState):
    """
    State to count discrete events.

    Args:
        name (str): Name of the state
        vmin (float): The allowed minimal value the state accepts
        vmax (float): The allowed maximal value the state accepts
        is_actionable (bool): If `True`, state can be changed by a policy
        is_observable (bool): If `True`, state can be observed by a policy
        exception_on_nan (bool): If `True` an exception is raised whenever an instance tries to
            access the value of the state and this value is `None`.
    """
    def __init__(
        self,
        name,
        vmin=0,
        vmax=np.inf,
        is_actionable=False,
        is_observable=True,
        exception_on_nan=False,
    ):
        super().__init__(
            name=name,
            is_actionable=is_actionable,
            is_observable=is_observable,
            vmin=vmin,
            vmax=vmax,
            exception_on_nan=exception_on_nan,
        )

    def assert_valid(self, value):
        NumericState.assert_valid(self, value)
        assert int(value) == value, f"Value {value} is not integer"

Data

Bases: object

Source code in lineflow/simulation/states.py

class Data(object):
    def __init__(self, feature_names, observables=None):
        self.feature_names = feature_names

        if observables is None:
            # All features are observable
            observables = (len(self.feature_names))*[True]
        self.observables = np.array(observables)

        self.T = np.array([])
        self.X = np.array([]).reshape(0, len(feature_names))

        self.modes = []
        for feature in self.feature_names:
            if feature.endswith('mode'):
                self.modes.append(True)
            else:
                self.modes.append(False)

    def append(self, end_time, values):
        self.T = np.append(self.T, end_time)
        self.X = np.vstack([self.X, values])

    def get_modes(self, lookback=None):
        """
        Returns the percent of working mode of the cells of a line over the
        lookback period
        """
        if lookback is None:
            lookback = self.T.shape[0]

        return self.X[-lookback:, self.modes]

    def get_uptime(self, lookback=None):
        """
        Returns the percentage of the station being in working mode
        (mode=0) over the lookback period
        """
        modes = self.get_modes(lookback=lookback)
        uptimes = (modes == 0).mean(axis=0)
        return uptimes

    def get_observations(self, lookback=None, include_time=True):
        """
        Here, only observable values are returned
        """
        if lookback is None:
            lookback = self.T.shape[0]

        X = self.X[-lookback:, self.observables]

        if include_time:
            T = self.T[-lookback:].reshape(-1, 1)
            T = T - T.max()
            return np.hstack([X, T])
        else:
            return X

    def df(self):
        df = pd.DataFrame(
            data=self.X,
            columns=self.feature_names,
        )
        df['T_end'] = self.T
        return df

get_modes(lookback=None)

Returns the percent of working mode of the cells of a line over the lookback period

Source code in lineflow/simulation/states.py

def get_modes(self, lookback=None):
    """
    Returns the percent of working mode of the cells of a line over the
    lookback period
    """
    if lookback is None:
        lookback = self.T.shape[0]

    return self.X[-lookback:, self.modes]

get_observations(lookback=None, include_time=True)

Here, only observable values are returned

Source code in lineflow/simulation/states.py

def get_observations(self, lookback=None, include_time=True):
    """
    Here, only observable values are returned
    """
    if lookback is None:
        lookback = self.T.shape[0]

    X = self.X[-lookback:, self.observables]

    if include_time:
        T = self.T[-lookback:].reshape(-1, 1)
        T = T - T.max()
        return np.hstack([X, T])
    else:
        return X

get_uptime(lookback=None)

Returns the percentage of the station being in working mode (mode=0) over the lookback period

Source code in lineflow/simulation/states.py

def get_uptime(self, lookback=None):
    """
    Returns the percentage of the station being in working mode
    (mode=0) over the lookback period
    """
    modes = self.get_modes(lookback=lookback)
    uptimes = (modes == 0).mean(axis=0)
    return uptimes

DiscreteState

Bases: BaseState

State to handle discrete states, like integer numbers or categories.

Parameters:

Name	Type	Description	Default
name	str	Name of the state	required
categories	list	List of values this state can take.	required
is_actionable	bool	If True, state can be changed by a policy	False
is_observable	bool	If True, state can be observed by a policy	True
exception_on_nan	bool	If True an exception is raised whenever an instance tries to access the value of the state and this value is None.	False

Source code in lineflow/simulation/states.py

class DiscreteState(BaseState):
    """
    State to handle discrete states, like integer numbers or categories.

    Args:
        name (str): Name of the state
        categories (list): List of values this state can take.
        is_actionable (bool): If `True`, state can be changed by a policy
        is_observable (bool): If `True`, state can be observed by a policy
        exception_on_nan (bool): If `True` an exception is raised whenever an instance tries to
            access the value of the state and this value is `None`.
    """

    def __init__(
        self,
        name,
        categories,
        is_actionable=False,
        is_observable=True,
        exception_on_nan=False,
    ):
        super().__init__(
            name=name,
            is_actionable=is_actionable,
            is_observable=is_observable,
            exception_on_nan=exception_on_nan,
        )
        self.categories = categories
        self.n_categories = len(self.categories)
        self.values = np.arange(self.n_categories)

        self._mapping = dict(zip(self.categories, self.values))

    def update(self, value):
        mapped_value = self._mapping[value]
        self._change_value(mapped_value)

    def to_str(self):
        return self.categories[self.value]

    def reverse(self, series):
        return series.astype(int).apply(lambda i: self.categories[i])

    def assert_valid(self, value):
        assert not isinstance(value, bool), f"{value} should not be boolean, but one of {self.values}"
        assert value in self.values, f"{value}, not in {self.values}"

    def set_next(self):
        self._change_value((self.value+1) % self.n_categories)

LineStates

Bases: object

Bag of all ObjectStates of all LineObjectss of a line.

Parameters:

Name	Type	Description	Default
objects	dict	Dict where keys are the object name and the value are of type ObjectStates.	required

Source code in lineflow/simulation/states.py

class LineStates(object):
    """
    Bag of all ObjectStates of all [`LineObjects`][lineflow.simulation.states.LineStates]s of a line.

    Args:
        objects (dict): Dict where keys are the object name and the value are of type
            [`ObjectStates`][lineflow.simulation.states.ObjectStates].
    """

    def __init__(self, objects: dict, env):
        self.objects = objects
        self.env = env

        # Fix an ordering of the objects
        self.object_names = [
            name for name in self.objects.keys()
        ]

        # Fix an order of the features
        self.feature_names = []
        self.observables = []
        self.actionables = []
        for name in self.object_names:
            self.feature_names.extend(self.objects[name]._get_names_with_prefix(name))
            self.observables.extend(self.objects[name].observables)
            self.actionables.extend(self.objects[name].actionables)

        self.data = Data(
            feature_names=self.feature_names,
            observables=self.observables)

    @property
    def observable_features(self):
        return [f for f, o in zip(self.feature_names, self.observables) if o]

    @property
    def actionable_features(self):
        return [f for f, a in zip(self.feature_names, self.actionables) if a]

    def get_actions(self):
        """
        Returns a list of actions for policies to design valid outputs
        """
        actions = {}

        for object_name in self.object_names:

            object_states = []
            for state_name in self[object_name].names:
                state = self[object_name][state_name]
                if state.is_actionable:
                    object_states.append(state)

            if len(object_states) > 0:
                actions[object_name] = object_states
        return actions

    def __getitem__(self, name):
        return self.objects[name]

    def apply(self, values):
        for object_name, object_values in values.items():
            self[object_name].apply(object_values)

    def update(self, values):
        for object_name, object_values in values.items():
            self[object_name].update(object_values)

    @property
    def values(self):
        data = np.array([], dtype=np.float32)
        for name in self.object_names:
            data = np.append(data, self.objects[name].values)
        return data

    def log(self):
        """
        Appends the (current) values of all objects to the data class
        """

        self.data.append(
            end_time=self.env.now,
            values=self.values
        )

    def get_observations(self, lookback=None, include_time=True):
        return self.data.get_observations(lookback=lookback, include_time=include_time)

    def get_n_parts_produced(self):
        return int(sum(
            [v for f, v in self.to_dict().items() if f.endswith('n_parts_produced')]
        ))

    def get_n_scrap_parts(self):
        return int(sum(
            [v for f, v in self.to_dict().items() if f.endswith('n_scrap_parts')]
        ))

    def get_uptime(self, lookback=None):
        return self.data.get_uptime(lookback=lookback)

    def to_dict(self):
        return dict(zip(self.feature_names, self.values))

    def __iter__(self):
        for object_name in self.object_names:
            for state_name in self[object_name].names:
                yield object_name, state_name

    def df(self, reverse=True, lookback=None):
        """
        This function is expensive in time and should only be called after simulation is finished
        """
        df = self.data.df()

        if lookback is not None:
            df = df.iloc[-lookback:]

        if reverse:
            for object_name, state_name in self:
                state = self[object_name][state_name]
                feature = f"{object_name}_{state_name}"
                if isinstance(state, DiscreteState) or isinstance(state, TokenState):
                    df[feature] = state.reverse(df[feature])

        df['T_start'] = df['T_end'].shift(1).fillna(0.0)
        return df

df(reverse=True, lookback=None)

This function is expensive in time and should only be called after simulation is finished

Source code in lineflow/simulation/states.py

def df(self, reverse=True, lookback=None):
    """
    This function is expensive in time and should only be called after simulation is finished
    """
    df = self.data.df()

    if lookback is not None:
        df = df.iloc[-lookback:]

    if reverse:
        for object_name, state_name in self:
            state = self[object_name][state_name]
            feature = f"{object_name}_{state_name}"
            if isinstance(state, DiscreteState) or isinstance(state, TokenState):
                df[feature] = state.reverse(df[feature])

    df['T_start'] = df['T_end'].shift(1).fillna(0.0)
    return df

get_actions()

Returns a list of actions for policies to design valid outputs

Source code in lineflow/simulation/states.py

def get_actions(self):
    """
    Returns a list of actions for policies to design valid outputs
    """
    actions = {}

    for object_name in self.object_names:

        object_states = []
        for state_name in self[object_name].names:
            state = self[object_name][state_name]
            if state.is_actionable:
                object_states.append(state)

        if len(object_states) > 0:
            actions[object_name] = object_states
    return actions

log()

Appends the (current) values of all objects to the data class

Source code in lineflow/simulation/states.py

def log(self):
    """
    Appends the (current) values of all objects to the data class
    """

    self.data.append(
        end_time=self.env.now,
        values=self.values
    )

NumericState

Bases: BaseState

State to handle numeric values.

Parameters:

Name	Type	Description	Default
name	str	Name of the state	required
vmin	float	The allowed minimal value the state accepts	-inf
vmax	float	The allowed maximal value the state accepts	inf
is_actionable	bool	If True, state can be changed by a policy	False
is_observable	bool	If True, state can be observed by a policy	True
exception_on_nan	bool	If True an exception is raised whenever an instance tries to access the value of the state and this value is None.	False

Source code in lineflow/simulation/states.py

class NumericState(BaseState):
    """
    State to handle numeric values.

    Args:
        name (str): Name of the state
        vmin (float): The allowed minimal value the state accepts
        vmax (float): The allowed maximal value the state accepts
        is_actionable (bool): If `True`, state can be changed by a policy
        is_observable (bool): If `True`, state can be observed by a policy
        exception_on_nan (bool): If `True` an exception is raised whenever an instance tries to
            access the value of the state and this value is `None`.
    """
    def __init__(
        self,
        name,
        vmin=-np.inf,
        vmax=np.inf,
        is_actionable=False,
        is_observable=True,
        exception_on_nan=False,
    ):
        super().__init__(
            name=name,
            is_actionable=is_actionable,
            is_observable=is_observable,
            exception_on_nan=exception_on_nan,
        )
        self.vmin = vmin
        self.vmax = vmax

    def assert_valid(self, value):
        assert (
            (self.vmin <= value) and
            (value <= self.vmax)
        ), f'Violated: {self.vmin}<={value}<={self.vmax}'

    def increment(self):
        """
        Increments the value by 1
        """
        self._change_value(self.value+1)

    def decrement(self):
        """
        Decrements the value by 1
        """
        self._change_value(self.value-1)

decrement()

Decrements the value by 1

Source code in lineflow/simulation/states.py

def decrement(self):
    """
    Decrements the value by 1
    """
    self._change_value(self.value-1)

increment()

Increments the value by 1

Source code in lineflow/simulation/states.py

def increment(self):
    """
    Increments the value by 1
    """
    self._change_value(self.value+1)

ObjectStates

Bases: object

Bag of all states of a LineObject

Parameters:

Name	Type	Description	Default
states	list	List of BaseState objects.	()

Source code in lineflow/simulation/states.py

class ObjectStates(object):
    """
    Bag of all states of a LineObject

    Args:
        states (list): List of [`BaseState`][lineflow.simulation.states.BaseState] objects.
    """

    def __init__(self, *states):
        self.names = [s.name for s in states]
        self.states = {
            state.name: state for state in states
        }

        self.observables = [self.states[n].is_observable for n in self.names]
        self.actionables = [self.states[n].is_actionable for n in self.names]

    def apply(self, values):
        """
        Applies the values to all states

        Args:
            values (dict): Dict where the keys are the names of the internal states and the values
                are the values to be applied.
        """
        for name, value in values.items():
            self[name].apply(value)

    def update(self, values):
        """
        Updates the values of all states

        Args:
            values (dict): Dict where the keys are the names of the internal states and the values
                are the values to be updated.
        """
        # TODO: Ugly code duplicate
        for name, value in values.items():
            self[name].update(value)

    @property
    def values(self):
        return np.array([self.states[n].value for n in self.names])

    def __getitem__(self, name):
        return self.states[name]

    def _get_names_with_prefix(self, prefix=None):
        prefix = "" if prefix is None else f"{prefix}_"
        return [f"{prefix}{n}" for n in self.names]

    def to_dict(self, prefix=None):
        return dict(
            zip(
                self._get_names_with_prefix(prefix),
                self.values
            )
        )

apply(values)

Applies the values to all states

Parameters:

Name	Type	Description	Default
values	dict	Dict where the keys are the names of the internal states and the values are the values to be applied.	required

Source code in lineflow/simulation/states.py

def apply(self, values):
    """
    Applies the values to all states

    Args:
        values (dict): Dict where the keys are the names of the internal states and the values
            are the values to be applied.
    """
    for name, value in values.items():
        self[name].apply(value)

update(values)

Updates the values of all states

Parameters:

Name	Type	Description	Default
values	dict	Dict where the keys are the names of the internal states and the values are the values to be updated.	required

Source code in lineflow/simulation/states.py

def update(self, values):
    """
    Updates the values of all states

    Args:
        values (dict): Dict where the keys are the names of the internal states and the values
            are the values to be updated.
    """
    # TODO: Ugly code duplicate
    for name, value in values.items():
        self[name].update(value)

TokenState

Bases: BaseState

State to handle discrete objects where its not clear from the begining which and how many objects need to be tracked.

Source code in lineflow/simulation/states.py

class TokenState(BaseState):
    """
    State to handle discrete objects where its not clear from the begining which and how many
    objects need to be tracked.
    """
    def __init__(self, name, mapping=None, is_actionable=False, is_observable=False):
        super().__init__(name=name, is_actionable=is_actionable, is_observable=is_observable)

        if mapping is None:
            self._mapping = {}
        else:
            self._mapping = mapping
        self.tokens = []

    def _get_next_value(self):

        if len(self._mapping.values()) == 0:
            return 0
        else:
            return max(self._mapping.values())+1

    def assert_valid(self, value):
        assert value in self._mapping.values()

    def update(self, token):
        if token not in self._mapping:
            # Generate new id for this token
            self._mapping[token] = self._get_next_value()
            self.tokens.append(token)

        value = self._mapping[token]
        self._change_value(value)

    def reverse(self, series):
        return series.astype(int).apply(lambda i: self.tokens[i])

Simulation

Line

Parameters:

Name	Type	Description	Default
realtime	bool	Only if visualize is True	False
factor	float	visualization speed	0.5
info	list	A list of line data that is retrivable over the get_info() method. That is info = [("A1", n_workers), ("A3", "assembly_time")]. Data will be logged in experiments.	None

Source code in lineflow/simulation/line.py

class Line:
    """
    Args:
        realtime (bool): Only if `visualize` is `True`
        factor (float): visualization speed
        info (list): A list of line data that is retrivable over the get_info() method.
            That is `info = [("A1", n_workers), ("A3", "assembly_time")]`.
            Data will be logged in experiments.
    """

    def __init__(
        self,
        realtime=False,
        factor=0.5,
        random_state=10,
        step_size=1,
        scrap_factor=1,
        info=None,
    ):

        # TODO: This attribute needs to be refactored in future as it is only used by the
        # gym-simulation
        self.scrap_factor = scrap_factor
        self.realtime = realtime
        self.factor = factor
        self.step_size = step_size
        if info is None:
            info = []
        self._info = info

        self.reset(random_state=random_state)

    @property
    def name(self):
        return self.__class__.__name__

    def info(self):
        """
        Returns additional Information about the line
        """
        general = {
            "name": self.name,
            "T": self.env.now,
            "n_parts": self.get_n_parts_produced(),
            "n_scrap_parts": self.get_n_scrap_parts(),
        }

        additional = {
            f"{station}_{attribute}": self.state.objects[station].states[attribute].value
            for station, attribute in self._info
        }
        return {**general, **additional}

    def _make_env(self):
        if self.realtime:
            self.env = simpy.rt.RealtimeEnvironment(factor=self.factor, strict=False)
        else:
            self.env = simpy.Environment()

    def _make_objects(self):
        """
        Builds the LineObjects
        """
        # Build the stations and connectors
        with StationaryObject() as objects:
            self.build()

        self._objects = {}

        for obj in objects:
            if obj.name in self._objects:
                raise ValueError(f'Multiple objects with name {obj.name} exist')
            self._objects[obj.name] = obj

        # Validate carrier specs
        for obj in self._objects.values():
            if hasattr(obj, 'carrier_specs'):
                self._validate_carrier_specs(obj.carrier_specs)

    def _validate_carrier_specs(self, specs):
        for carrier_name, part_specs in specs.items():
            for part_name, part_spec in part_specs.items():
                for station in part_spec.keys():
                    if station not in self._objects:
                        raise ValueError(
                                f"Spec for part '{part_name}' in carrier '{carrier_name}' "
                                f"contains unkown station '{station}'"
                        )

    def _build_states(self):
        """
        Builds the states of the line objects as well as the LineState
        """
        object_states = {}

        for name, obj in self._objects.items():
            obj.init(self.random)
            object_states[name] = obj.state

        self.state = LineStates(object_states, self.env)

    def reset(self, random_state=None):
        """
        Resets the simulation.
        """
        self.random = np.random.RandomState(random_state)
        self._make_env()
        self._make_objects()

        self._build_states()
        self._register_objects_at_env()

        self.end_step = 0
        self.env.process(self.step_event())

    def _assert_one_sink(self):
        if len([c for c in self._objects.values() if isinstance(c, Sink)]) != 1:
            raise ValueError(
                "Number of sinks does not match"
                "Currently, only scenarios with exactly one sink are allowed"
            )

    def get_sink(self):
        sinks = [s for s in self._objects.values() if isinstance(s, Sink)]
        self._assert_one_sink()
        return sinks[0]

    def get_n_scrap_parts(self):
        """
        Returns the number of produced parts up to now
        """
        return self.state.get_n_scrap_parts()

    def get_n_parts_produced(self):
        """
        Returns the number of produced parts up to now
        """
        return self.state.get_n_parts_produced()

    def get_uptime(self, lookback=None):
        """
        Returns the uptime of the line 
        """
        return self.state.get_uptime(lookback=lookback)

    def build(self):
        """
        This function should add objects of the LineObject class as attributes
        """
        raise NotImplementedError()

    def _register_objects_at_env(self):
        """
        Registers all line objects at the simpy simulation environment.
        """
        for o in self._objects.values():
            o.register(self.env)

    def _draw(self, actions=None):

        self.viewpoint.check_user_input()

        self.viewpoint.clear()

        # Draw objects, first connectors, then stations
        self._draw_connectors()
        self._draw_stations()

        self.viewpoint._draw()

        if actions is not None:
            self._draw_actions(actions)

        self._draw_info()

        pygame.display.flip()

    def _draw_info(self):

        font = pygame.font.SysFont(None, 20)

        time = font.render('T={:.2f}'.format(self.env.now), True, 'black')
        n_parts = font.render(
            f'#Parts={self.get_n_parts_produced()}', True, 'black'
        )
        self.viewpoint.screen.blit(time, time.get_rect(center=(30, 30)))
        self.viewpoint.screen.blit(n_parts, n_parts.get_rect(center=(30, 50)))

    def _draw_actions(self, actions):
        font = pygame.font.SysFont(None, 20)
        actions = font.render(f'{actions}', True, 'black')
        self.viewpoint.screen.blit(actions, actions.get_rect(center=(500, 30)))

    def _draw_stations(self):
        self._draw_objects_of_type(Station)

    def _draw_connectors(self):
        self._draw_objects_of_type(Connector)

    def _draw_objects_of_type(self, object_type):
        for _, obj in self._objects.items():
            if isinstance(obj, object_type):
                obj._draw(self.viewpoint.paper)

    def _get_object_positions(self):
        x = []
        y = []
        for o in self._objects.values():
            if hasattr(o, "position"):
                x.append(o.position[0])
                y.append(o.position[1])
        return x, y

    def _adjust_positions(self):
        x, y = self._get_object_positions()

        if min(x) < 100:
            delta_x = 100 - min(x)
            for o in self._objects.values():
                if hasattr(o, "position"):
                    o.position[0] += delta_x
        if min(y) < 100:
            delta_y = 100 - min(y)
            for o in self._objects.values():
                if hasattr(o, "position"):
                    o.position[1] += delta_y

        x, y = self._get_object_positions()
        return max(x), max(y)

    def setup_draw(self):
        pygame.init()

        max_x, max_y = self._adjust_positions()
        for o in self._objects.values():
            o.setup_draw()

        self.viewpoint = Viewpoint(size=(max_x+100, max_y+100))

    def apply(self, values):
        for object_name in values.keys():
            self._objects[object_name].apply(values[object_name])

    def step(self, simulation_end=None):
        """
        Step to the next state of the line
        Args:
            simulation_end (int):
                Time until terminated flag is returned as True. If None
                terminated is always False.
        """
        terminated = False

        # The end of the the current step, excluding the event execution
        # i.e. execute all events where scheudled_time < end_step
        self.end_step = self.end_step + self.step_size

        while True:
            if self.env.peek() > self.end_step:
                self.state.log()
                # If the next event is scheduled after simulation end
                if simulation_end is not None and self.env.peek() > simulation_end:
                    terminated = True

                return self.state, terminated

            self.env.step()

    def step_event(self):
        """
        Ensures that there is an Event scheduled for `self.step_size` intervals
        The step function is only able to stop the simulation if an Event is scheduled.
        """
        while True:
            yield self.env.timeout(self.step_size)

    def run(
        self,
        simulation_end,
        agent=None,
        show_status=True,
        visualize=False,
        capture_screen=False,
    ):
        """
        Args:
            simulation_end (float): Time until the simulation stops
            agent (lineflow.models.reinforcement_learning.agents): An Agent that interacts with a
                line. Can also be just a policy if an __call__ method exists like in the BaseAgent
                class.
            show_status (bool): Show progress bar for each simulation episode
            visualize (bool): If true, line visualization is opened
            capture_screen (bool): Captures last Time frame when screen should be recorded
        """

        if visualize:
            self.setup_draw()


        # Register objects when simulation is initially started
        if len(self.env._queue) == 0:
            self._register_objects_at_env()

        now = 0
        actions = None
        pbar = tqdm(
            total=simulation_end,
            bar_format='{desc}: {percentage:3.2f}%|{bar:50}|',
            disable=not show_status,
        )

        while self.env.now < simulation_end:
            pbar.update(self.env.now - now)
            now = self.env.now
            try:
                self.step()
            except simpy.core.EmptySchedule:
                logger.warning('Simulation in dead-lock - end early')
                break

            if agent is not None:
                actions = agent(self.state, self.env)
                self.apply(actions)

            if visualize:
                self._draw(actions)

        if capture_screen and visualize:
            pygame.image.save(self.viewpoint.screen, f"{self.name}.png")

        if visualize:
            self.viewpoint.teardown()

    def get_observations(self, object_name=None):
        """
        """

        df = self.state.df()

        if object_name is None:
            return df
        else:
            cols = [c for c in df.columns if c.startswith(object_name)]
            cols = cols + ['T_start', 'T_end']
            return df[cols].rename(
                columns={
                    c: c.replace(object_name + '_', '') for c in cols
                }
            )

    def __getitem__(self, name):
        return self._objects[name]

build()

This function should add objects of the LineObject class as attributes

Source code in lineflow/simulation/line.py

def build(self):
    """
    This function should add objects of the LineObject class as attributes
    """
    raise NotImplementedError()

get_n_parts_produced()

Returns the number of produced parts up to now

Source code in lineflow/simulation/line.py

def get_n_parts_produced(self):
    """
    Returns the number of produced parts up to now
    """
    return self.state.get_n_parts_produced()

get_n_scrap_parts()

Returns the number of produced parts up to now

Source code in lineflow/simulation/line.py

def get_n_scrap_parts(self):
    """
    Returns the number of produced parts up to now
    """
    return self.state.get_n_scrap_parts()

get_observations(object_name=None)

Source code in lineflow/simulation/line.py

def get_observations(self, object_name=None):
    """
    """

    df = self.state.df()

    if object_name is None:
        return df
    else:
        cols = [c for c in df.columns if c.startswith(object_name)]
        cols = cols + ['T_start', 'T_end']
        return df[cols].rename(
            columns={
                c: c.replace(object_name + '_', '') for c in cols
            }
        )

get_uptime(lookback=None)

Returns the uptime of the line

Source code in lineflow/simulation/line.py

def get_uptime(self, lookback=None):
    """
    Returns the uptime of the line 
    """
    return self.state.get_uptime(lookback=lookback)

info()

Returns additional Information about the line

Source code in lineflow/simulation/line.py

def info(self):
    """
    Returns additional Information about the line
    """
    general = {
        "name": self.name,
        "T": self.env.now,
        "n_parts": self.get_n_parts_produced(),
        "n_scrap_parts": self.get_n_scrap_parts(),
    }

    additional = {
        f"{station}_{attribute}": self.state.objects[station].states[attribute].value
        for station, attribute in self._info
    }
    return {**general, **additional}

reset(random_state=None)

Resets the simulation.

Source code in lineflow/simulation/line.py

def reset(self, random_state=None):
    """
    Resets the simulation.
    """
    self.random = np.random.RandomState(random_state)
    self._make_env()
    self._make_objects()

    self._build_states()
    self._register_objects_at_env()

    self.end_step = 0
    self.env.process(self.step_event())

run(simulation_end, agent=None, show_status=True, visualize=False, capture_screen=False)

Parameters:

Name	Type	Description	Default
simulation_end	float	Time until the simulation stops	required
agent	agents	An Agent that interacts with a line. Can also be just a policy if an call method exists like in the BaseAgent class.	None
show_status	bool	Show progress bar for each simulation episode	True
visualize	bool	If true, line visualization is opened	False
capture_screen	bool	Captures last Time frame when screen should be recorded	False

Source code in lineflow/simulation/line.py

def run(
    self,
    simulation_end,
    agent=None,
    show_status=True,
    visualize=False,
    capture_screen=False,
):
    """
    Args:
        simulation_end (float): Time until the simulation stops
        agent (lineflow.models.reinforcement_learning.agents): An Agent that interacts with a
            line. Can also be just a policy if an __call__ method exists like in the BaseAgent
            class.
        show_status (bool): Show progress bar for each simulation episode
        visualize (bool): If true, line visualization is opened
        capture_screen (bool): Captures last Time frame when screen should be recorded
    """

    if visualize:
        self.setup_draw()


    # Register objects when simulation is initially started
    if len(self.env._queue) == 0:
        self._register_objects_at_env()

    now = 0
    actions = None
    pbar = tqdm(
        total=simulation_end,
        bar_format='{desc}: {percentage:3.2f}%|{bar:50}|',
        disable=not show_status,
    )

    while self.env.now < simulation_end:
        pbar.update(self.env.now - now)
        now = self.env.now
        try:
            self.step()
        except simpy.core.EmptySchedule:
            logger.warning('Simulation in dead-lock - end early')
            break

        if agent is not None:
            actions = agent(self.state, self.env)
            self.apply(actions)

        if visualize:
            self._draw(actions)

    if capture_screen and visualize:
        pygame.image.save(self.viewpoint.screen, f"{self.name}.png")

    if visualize:
        self.viewpoint.teardown()

step(simulation_end=None)

Step to the next state of the line Args: simulation_end (int): Time until terminated flag is returned as True. If None terminated is always False.

Source code in lineflow/simulation/line.py

def step(self, simulation_end=None):
    """
    Step to the next state of the line
    Args:
        simulation_end (int):
            Time until terminated flag is returned as True. If None
            terminated is always False.
    """
    terminated = False

    # The end of the the current step, excluding the event execution
    # i.e. execute all events where scheudled_time < end_step
    self.end_step = self.end_step + self.step_size

    while True:
        if self.env.peek() > self.end_step:
            self.state.log()
            # If the next event is scheduled after simulation end
            if simulation_end is not None and self.env.peek() > simulation_end:
                terminated = True

            return self.state, terminated

        self.env.step()

step_event()

Ensures that there is an Event scheduled for self.step_size intervals The step function is only able to stop the simulation if an Event is scheduled.

Source code in lineflow/simulation/line.py

def step_event(self):
    """
    Ensures that there is an Event scheduled for `self.step_size` intervals
    The step function is only able to stop the simulation if an Event is scheduled.
    """
    while True:
        yield self.env.timeout(self.step_size)

Assembly

Bases: Station

Assembly takes a carrier from buffer_in and buffer_component, puts the parts of the component carrier on the carrier that came from buffer_in, and pushes that carrier to buffer_out and pushes the component carrier to buffer_return if a buffer return exists, otherwise these carriers are lost.

Parameters:

Name	Type	Description	Default
name	str	Name of the station	required
processing_time	float	Time until parts are moved from component carrier to main carrier	5
position	tuple	X and Y position in the visualization	None
buffer_return	Buffer	The buffer to put the old component carriers on	None
processing_std	float	The standard deviation of the processing time	None

Source code in lineflow/simulation/stations.py

class Assembly(Station):
    """
    Assembly takes a carrier from `buffer_in` and `buffer_component`, puts the parts of the component
    carrier on the carrier that came from buffer_in, and pushes that carrier to buffer_out and
    pushes the component carrier to buffer_return if a buffer return exists, otherwise these
    carriers are lost.

    Args:
        name (str): Name of the station
        processing_time (float): Time until parts are moved from component carrier to main carrier
        position (tuple): X and Y position in the visualization
        buffer_return (lineflow.simulation.connectors.Buffer): The buffer to
            put the old component carriers on
        processing_std (float): The standard deviation of the processing time
    """
    def __init__(
        self,
        name,
        buffer_in=None,
        buffer_out=None,
        buffer_component=None,
        buffer_return=None,
        processing_time=5,
        position=None,
        processing_std=None,
        NOK_part_error_time=2,
        worker_pool=None,
    ):

        super().__init__(
            name=name,
            position=position,
            processing_time=processing_time,
            processing_std=processing_std,
            worker_pool=worker_pool,
        )
        self.NOK_part_error_time = NOK_part_error_time

        if buffer_in is not None:
            self._connect_to_input(buffer_in)

        if buffer_out is not None:
            self._connect_to_output(buffer_out)

        self.buffer_component = []

        if buffer_component is not None:

            if isinstance(buffer_component, Buffer):
                buffer_component = [buffer_component]

            for buffer in buffer_component:
                self._connect_to_component_input(buffer)

        if buffer_return is not None:
            self.buffer_return = buffer_return.connect_to_input(self)

    def init_state(self):
        self.state = ObjectStates(
            DiscreteState('on', categories=[True, False], is_actionable=False, is_observable=False),
            DiscreteState('mode', categories=['working', 'waiting', 'failing']),
            TokenState(name='carrier', is_observable=False),
            TokenState(name='carrier_component', is_observable=False),
            CountState('n_scrap_parts', is_actionable=False, is_observable=True),
            CountState('n_workers', is_actionable=False, is_observable=True, vmin=0),
            NumericState('processing_time', is_actionable=False, is_observable=True, vmin=0),
        )
        self.state['on'].update(True)
        self.state['mode'].update("waiting")
        self.state['carrier'].update(None)
        self.state['carrier_component'].update(None)
        self.state['n_scrap_parts'].update(0)
        self.state['processing_time'].update(self.processing_time)
        self.state['n_workers'].update(self.n_workers)

    def connect_to_component_input(self, station, *args, **kwargs):
        buffer = Buffer(name=f"Buffer_{station.name}_to_{self.name}", *args, **kwargs)
        self._connect_to_component_input(buffer)
        station._connect_to_output(buffer)

    def connect_to_component_return(self, station, *args, **kwargs):
        buffer = Buffer(name=f"Buffer_{self.name}_to_{station.name}", *args, **kwargs)
        self.buffer_return = buffer.connect_to_input(self)
        station._connect_to_input(buffer)

    def _connect_to_component_input(self, buffer):
        self.buffer_component.append(buffer.connect_to_output(self))

    def _has_invalid_components_on_carrier(self, carrier):
        """
        Checks if any of the components on the carrier is not valid for assembly. In this case,
        `True` is returned. Otherwise, `False` is returned.
        """
        for component in carrier:
            # Check assembly condition
            if not component.is_valid_for_assembly(self.name):
                return True

            # Check if part is self._is_nok_part(component):
            if self._is_nok_part(component):
                return True
        return False

    def _draw_info(self, screen):
        self._draw_n_workers(screen)

    def run(self):

        while True:
            if self.is_on():

                yield self.env.process(self.request_workers())
                self.state['n_workers'].update(self.n_workers)
                # Wait to get part from buffer_in
                yield self.env.process(self.set_to_waiting())
                carrier = yield self.env.process(self.buffer_in())

                # Update current_carrier and count parts of carrier
                self.state['carrier'].update(carrier.name)


                carrier_components = []
                for buffer_component in self.buffer_component:
                    # Run until carrier with components each having a valid assembly condition is
                    # received
                    while True:
                        # Wait to get component
                        carrier_component = yield self.env.process(buffer_component())
                        self.state['carrier_component'].update(carrier_component.name)

                        # Check component
                        if self._has_invalid_components_on_carrier(carrier_component):
                            yield self.env.process(self.set_to_error())
                            yield self.env.timeout(self.NOK_part_error_time)
                            self.state['n_scrap_parts'].increment()

                            # send carrier back
                            if hasattr(self, 'buffer_return'):
                                carrier_component.parts.clear()
                                yield self.env.process(self.buffer_return(carrier_component))
                            yield self.env.process(self.set_to_waiting())
                            continue

                        else:
                            # All components are valid, proceed with assembly
                            carrier_components.append(carrier_component)
                            break

                # Process components
                yield self.env.process(self.set_to_work())

                add_time = sum(
                    [c.get_additional_processing_time(self.name) for c in carrier_components]
                )
                processing_time = self._sample_exp_time(
                    time=self.processing_time + add_time,
                    scale=self.processing_std,
                )
                yield self.env.timeout(processing_time)
                self.state['processing_time'].update(processing_time)

                for carrier_component in carrier_components:
                    for component in carrier_component:
                        carrier.assemble(component)

                # Release workers
                self.release_workers()

                # Place carrier on buffer_out
                yield self.env.process(self.set_to_waiting())
                yield self.env.process(self.buffer_out(carrier))
                self.state['carrier'].update(None)

                # send component carrier back
                if hasattr(self, 'buffer_return'):
                    yield self.env.process(self.buffer_return(carrier_component))

                self.state['carrier_component'].update(None)

            else:
                yield self.turn_off()

Magazine

Bases: Station

Magazine station manages carriers.

The Magazine gets carriers from buffer_in and stores them in the magazine. Afterwards it takes a carrier from its magazine and pushes the carrier to buffer_out. If unlimited_carriers is True no buffer_in is needed.

Parameters:

Name	Type	Description	Default
unlimited_carriers	bool	If True, the Magazine will have an unlimited amount of carriers available	True
carriers_in_magazine	int	Number of carriers in the magazine	0
carrier_getting_time	float	Time to get a carrier from the magazine	2
actionable_magazine	bool	If True, carriers in the magazine is in an actionable state	True
carrier_capacity	int	Defines how many parts can be assembled on a carrier. If set to default (infinity) or > 15, carrier will be visualized with one part.	inf

Source code in lineflow/simulation/stations.py

class Magazine(Station):
    '''
    Magazine station manages carriers.

    The Magazine gets carriers from buffer_in and stores them in the
    magazine. Afterwards it takes a carrier from its magazine and pushes the
    carrier to buffer_out.
    If unlimited_carriers is True no buffer_in is needed.

    Args:
        unlimited_carriers (bool): If True, the Magazine will have an unlimited amount of carriers available
        carriers_in_magazine (int): Number of carriers in the magazine
        carrier_getting_time (float): Time to get a carrier from the magazine
        actionable_magazine (bool): If True, carriers in the magazine is in an actionable state
        carrier_capacity (int): Defines how many parts can be assembled on a carrier. If set to default (infinity) or > 15, carrier will be visualized with one part.
    '''
    def __init__(
        self,
        name,
        buffer_in=None,
        buffer_out=None,
        position=None,
        unlimited_carriers=True,
        carrier_capacity=np.inf,
        actionable_magazine=True,
        carrier_getting_time=2,
        carriers_in_magazine=0,
        carrier_min_creation=1,
        carrier_max_creation=None,
    ):
        super().__init__(
            name=name,
            position=position,
        )
        self._assert_init_args(buffer_in, unlimited_carriers, carriers_in_magazine, carrier_capacity)

        if buffer_in is not None:
            self._connect_to_input(buffer_in)
        if buffer_out is not None:
            self._connect_to_output(buffer_out)

        self.actionable_magazine = actionable_magazine
        self.init_carriers_in_magazine = carriers_in_magazine
        self.carrier_getting_time = carrier_getting_time

        self.unlimited_carriers = unlimited_carriers
        self.carrier_capacity = carrier_capacity
        self.carrier_id = 1
        self.carrier_min_creation = carrier_min_creation
        self.carrier_max_creation = carrier_max_creation if carrier_max_creation is not None else 2*carrier_min_creation
        self._carrier_counter = 0

    def init_state(self):

        self.state = ObjectStates(
            DiscreteState('on', categories=[True, False], is_actionable=False, is_observable=False),
            DiscreteState('mode', categories=['working', 'waiting', 'failing']),
            CountState('carriers_in_magazine', is_actionable=self.actionable_magazine, is_observable=True),
            TokenState(name='carrier', is_observable=False),
            TokenState(name='part', is_observable=False),
        )

        self.state['carriers_in_magazine'].update(self.init_carriers_in_magazine)
        self.state['on'].update(True)
        self.state['mode'].update("waiting")
        self.state['carrier'].update(None)
        self.state['part'].update(None)

    def _assert_init_args(self, buffer_in, unlimited_carriers, carriers_in_magazine, carrier_capacity):
        if carrier_capacity > 15:
            warnings.warn("If carrier_capacity > 15, visualization of parts on carriers is restriced and carrier will be visualized with one part")

        if not isinstance(carrier_capacity, int) and carrier_capacity != np.inf:
            raise AttributeError("Type of carrier capacity must be int or np.inf")

        if not unlimited_carriers and carriers_in_magazine == 0:
            raise AttributeError(f"unlimited_carriers is {unlimited_carriers} and cell also has 0 carriers in magazine")

        if unlimited_carriers and carriers_in_magazine > 0:
            raise AttributeError(f"unlimited_carriers is {unlimited_carriers} and cell has more than 0 carriers in magazine")

        if buffer_in and unlimited_carriers:
                raise AttributeError(f"Only magazine or unlimited_carriers {unlimited_carriers} is required")

    def create_carrier(self):
        if self._carrier_counter == 0:
            self._carrier_counter = self.random.randint(
                self.carrier_min_creation, 
                self.carrier_max_creation + 1,
            )

        name = f'{self.name}_{self.carrier_id}'
        carrier = Carrier(
            self.env, 
            name=name, 
            capacity=self.carrier_capacity, 
        )
        self.carrier_id += 1
        self._carrier_counter -= 1

        return carrier

    def _initial_fill_magazine(self, n_carriers):
        # attribute needs to be set here as env is not available in __init__()
        self.magazine = simpy.Store(self.env)
        for i in range(n_carriers):
            carrier = self.create_carrier()
            self.magazine.put(carrier)

    def get_carrier_from_magazine(self):
        yield self.env.process(self._update_magazine())
        yield self.env.timeout(self.carrier_getting_time)

        while True:
            yield self.env.process(self._update_magazine())
            yield self.env.process(self.set_to_work())
            if len(self.magazine.items) > 0:
                carrier = yield self.magazine.get()
                break
            else:
                yield self.env.process(self.set_to_waiting())
                yield self.env.timeout(1)

        self.state['carriers_in_magazine'].decrement()
        return carrier

    def _buffer_in_to_magazine(self):
        while True:
            carrier = yield self.env.process(self.buffer_in())
            yield self.env.process(self.add_carrier_to_magazine(carrier))

    def add_carrier_to_magazine(self, carrier):
        yield self.magazine.put(carrier)
        self.state['carriers_in_magazine'].increment()

    def _update_magazine(self):
        '''
        update the magazine according to state
        '''
        should = self.state['carriers_in_magazine'].value
        current = len(self.magazine.items)
        diff = should - current
        if diff > 0:
            for i in range(diff):
                carrier = self.create_carrier()
                self.magazine.put(carrier)

        if diff < 0:
            for i in range(abs(diff)):
                carrier = yield self.magazine.get()

    def _draw_info(self, screen):
        self._draw_n_carriers(screen)

    def get_carrier(self):
        # First check if Magazine is allowed to create unlimited carriers
        if self.unlimited_carriers:
            yield self.env.timeout(self.carrier_getting_time)
            carrier = self.create_carrier()

        # Second check magazine
        else:
            carrier = yield self.env.process(self.get_carrier_from_magazine())
        self.state["carrier"].update(carrier.name)
        return carrier

    def run(self):
        # Initially fill the magazine with carriers
        self._initial_fill_magazine(self.state['carriers_in_magazine'].value)

        if hasattr(self, 'buffer_in'):
            self.env.process(self._buffer_in_to_magazine())

        while True:
            if self.is_on():
                # Get carrier from Magazine
                yield self.env.process(self.set_to_work())
                carrier = yield self.env.process(self.get_carrier())

                # Wait to place carrier to buffer_out
                yield self.env.process(self.set_to_waiting())
                yield self.env.process(self.buffer_out(carrier))
                self.state['carrier'].update(None)
            else:
                yield self.turn_off()

Process

Bases: Station

Process stations take a carrier as input, process the carrier, and push it onto buffer_out

Parameters:

Name	Type	Description	Default
name	str	Name of the station	required
buffer_in	Buffer	Buffer in	None
buffer_out	Buffer	Buffer out	None
processing_time	float	Time it takes to process the carrier (carrier needs to be available)	5
processing_std	float	Standard deviation of the processing time	None
rework_probability	float	Probability of a carrier to be reworked (takes 2x the time)	0
position	tuple	X and Y position in visualization	None
worker_pool	WorkerPool	Worker pool to	None

Source code in lineflow/simulation/stations.py

class Process(Station):
    """
    Process stations take a carrier as input, process the carrier, and push it onto `buffer_out`

    Args:
        name (str): Name of the station
        buffer_in (lineflow.simulation.connectors.Buffer, optional): Buffer in
        buffer_out (lineflow.simulation.connectors.Buffer, optional): Buffer out
        processing_time (float): Time it takes to process the carrier (carrier needs to be
            available)
        processing_std (float): Standard deviation of the processing time
        rework_probability (float): Probability of a carrier to be reworked (takes 2x the time)
        position (tuple): X and Y position in visualization
        worker_pool (lineflow.simulation.stations.WorkerPool, optional): Worker pool to
    """

    def __init__(
        self,
        name,
        buffer_in=None,
        buffer_out=None,
        processing_time=5,
        position=None,
        processing_std=None,
        rework_probability=0,
        worker_pool=None,
    ):

        super().__init__(
            name=name,
            position=position,
            processing_time=processing_time,
            processing_std=processing_std,
            rework_probability=rework_probability,
            worker_pool=worker_pool,
        )

        if buffer_in is not None:
            self._connect_to_input(buffer_in)

        if buffer_out is not None:
            self._connect_to_output(buffer_out)

    def init_state(self):

        self.state = ObjectStates(
            DiscreteState('on', categories=[True, False], is_actionable=False, is_observable=False),
            DiscreteState('mode', categories=['working', 'waiting', 'failing']),
            TokenState(name='carrier', is_observable=False),
            NumericState('processing_time', is_actionable=False, is_observable=True, vmin=0),
            CountState('n_workers', is_actionable=False, is_observable=True, vmin=0),
        )
        self.state['on'].update(True)
        self.state['mode'].update("waiting")
        self.state['carrier'].update(None)
        self.state['processing_time'].update(self.processing_time)
        self.state['n_workers'].update(self.n_workers)

    def _draw_info(self, screen):
        self._draw_n_workers(screen)

    def run(self):

        while True:
            if self.is_on():
                yield self.env.process(self.request_workers())
                self.state['n_workers'].update(self.n_workers)
                # Wait to get part from buffer_in
                yield self.env.process(self.set_to_waiting())
                carrier = yield self.env.process(self.buffer_in())
                self.state['carrier'].update(carrier.name)

                yield self.env.process(self.set_to_work())

                processing_time = self._sample_exp_time(
                    time=self.processing_time + carrier.get_additional_processing_time(self.name),
                    scale=self.processing_std,
                    rework_probability=self.rework_probability,
                )
                yield self.env.timeout(processing_time)
                self.state['processing_time'].update(processing_time)

                # Release workers
                self.release_workers()

                # Wait to place carrier to buffer_out
                yield self.env.process(self.set_to_waiting())
                yield self.env.process(self.buffer_out(carrier))
                self.state['carrier'].update(None)

            else:
                yield self.turn_off()

SequentialProcess

Bases: Process

SequentialProcess stations take a carrier as input, process each part on the carrier sequentially, and push the carrier onto buffer_out. The processing time of each part can be different and is determined by the part specifications. If a part does not fulfill the assembly condition, it is scrapped and the carrier is reworked (takes 2x the time). If connected to a worker pool, workers are released and requested for each part on the carrier.

The processing time for each part is the sum of the processing time of the station (which is 0 by default) and the additional processing time specified in the part specifications for this station. The standard deviation of the processing time is determined by the processing_std attribute of the station and is applied to the total processing time of the part.

Parameters:

Name	Type	Description	Default
name	str	Name of the station	required
buffer_in	Buffer	Buffer in	None
buffer_out	Buffer	Buffer out	None
processing_time	float	Time it takes to process each part on the carrier	0
processing_std	float	Standard deviation of the processing time	0.1
position	tuple	X and Y position in visualization	None
worker_pool	WorkerPool	Worker pool to	None

Example:

Assume a source produces carriers of type Type_A with two parts Part1 and Part2 that have the following specifications:

carrier_specs = {
    'Type_A': {
        'Part1': {
            'Process1': {
                'extra_processing_time': 20,
                'error_probability': 0.0,
                'error_time': 0,
            }, 
            'Process2': {
                'extra_processing_time': 5,
                'error_probability': 0.0,
                'error_time': 0,
                }
        },
        'Part2': {
            'Process2': {
                'extra_processing_time': 8,
                'error_probability': 0.5,
                'error_time': 10,
            }
        }
    },
}

Assume the processes are initialized as follows:

p1 = SequentialProcess(
    name='Process1',
    processing_time=0,
    processing_std=0.1,
    rework_probability=0.5,
)

p2 = SequentialProcess(
    name='Process2',
    processing_time=10,
    processing_std=0.1,
    rework_probability=0.5,
)

Both processes process all parts.

At Process1, a carrier of type Type_A with a processing time of

\[ \underbrace{0 + 20}_{\mathrm{Part1}} + \underbrace{0 + 0}_{\mathrm{Part2}} = 20 \]

At Process2, a carrier of type Type_A with a processing time of

\[ \underbrace{10 + 5}_{\mathrm{Part1}} + \underbrace{10 + 8}_{\mathrm{Part2}} = 33 \]

Source code in lineflow/simulation/stations.py

class SequentialProcess(Process):
    """
    SequentialProcess stations take a carrier as input, process each part on the carrier
    sequentially, and push the carrier onto buffer_out. The processing time of each part can be
    different and is determined by the part specifications. If a part does not fulfill the assembly
    condition, it is scrapped and the carrier is reworked (takes 2x the time). If connected to a
    worker pool, workers are released and requested for each part on the carrier. 

    The processing time for each part is the sum of the processing time of the station (which is 0
    by default) and the additional processing time specified in the part specifications for this
    station. The standard deviation of the processing time is determined by the `processing_std`
    attribute of the station and is applied to the total processing time of the part.

    Args:
        name (str): Name of the station
        buffer_in (lineflow.simulation.connectors.Buffer, optional): Buffer in
        buffer_out (lineflow.simulation.connectors.Buffer, optional): Buffer out
        processing_time (float): Time it takes to process each part on the carrier 
        processing_std (float): Standard deviation of the processing time
        position (tuple): X and Y position in visualization
        worker_pool (lineflow.simulation.stations.WorkerPool, optional): Worker pool to


    Example:

    Assume a source produces carriers of type `Type_A` with two parts `Part1` and `Part2` that have the
    following specifications:

    ```python
    carrier_specs = {
        'Type_A': {
            'Part1': {
                'Process1': {
                    'extra_processing_time': 20,
                    'error_probability': 0.0,
                    'error_time': 0,
                }, 
                'Process2': {
                    'extra_processing_time': 5,
                    'error_probability': 0.0,
                    'error_time': 0,
                    }
            },
            'Part2': {
                'Process2': {
                    'extra_processing_time': 8,
                    'error_probability': 0.5,
                    'error_time': 10,
                }
            }
        },
    }
    ```

    Assume the processes are initialized as follows:

    ```python
    p1 = SequentialProcess(
        name='Process1',
        processing_time=0,
        processing_std=0.1,
        rework_probability=0.5,
    )

    p2 = SequentialProcess(
        name='Process2',
        processing_time=10,
        processing_std=0.1,
        rework_probability=0.5,
    )
    ```

    Both processes process all parts. 

    At `Process1`, a carrier of type `Type_A` with a processing time of 

    $$
    \\underbrace{0 + 20}_{\\mathrm{Part1}} + \\underbrace{0 + 0}_{\\mathrm{Part2}} = 20
    $$


    At `Process2`, a carrier of type `Type_A` with a processing time of 

    $$
    \\underbrace{10 + 5}_{\\mathrm{Part1}} + \\underbrace{10 + 8}_{\\mathrm{Part2}} = 33
    $$

    """
    def __init__(
        self,
        name,
        buffer_in=None,
        buffer_out=None,
        position=None,
        worker_pool=None,
        processing_time=0,
        processing_std=0.1,
        error_std=0,
    ):

        super().__init__(
            name=name,
            buffer_in=buffer_in,
            buffer_out=buffer_out,
            processing_time=processing_time,
            position=position,
            processing_std=processing_std,
            rework_probability=0,
            worker_pool=worker_pool,
        )

        self.processing_std = processing_std
        self.error_std = error_std
        assert self.processing_std >= 0 and self.processing_std <= 1
        assert self.error_std >= 0 and self.error_std <= 1

    def run(self):

        while True:
            if self.is_on():
                # Wait to get part from buffer_in
                yield self.env.process(self.set_to_waiting())
                carrier = yield self.env.process(self.buffer_in())
                self.state['carrier'].update(carrier.name)

                yield self.env.timeout(2)

                total_processing_time = 0

                for part in carrier.get_parts_for_station(self.name):

                    yield self.env.process(self.request_workers())
                    self.state['n_workers'].update(self.n_workers)

                    error_proba = part.get_error_probability(self.name)

                    if self.random.uniform(0, 1) < error_proba:
                        yield self.env.process(self.set_to_error())
                        # Release workers
                        self.release_workers()
                        self.state['n_workers'].update(self.n_workers)
                        error_time = self._sample_exp_time(
                            time=part.get_error_time(self.name),
                            scale=self.error_std,
                            rework_probability=self.rework_probability,
                        )

                        yield self.env.timeout(error_time)
                        total_processing_time += error_time
                        yield self.env.process(self.set_to_waiting())
                        yield self.env.process(self.request_workers())
                        self.state['n_workers'].update(self.n_workers)

                    yield self.env.process(self.set_to_work())

                    processing_time = self._sample_exp_time(
                        time=self.processing_time+part.get_processing_time(self.name),
                        scale=self.processing_std,
                        rework_probability=self.rework_probability,
                    )
                    yield self.env.timeout(processing_time)
                    total_processing_time += processing_time
                    # Release workers
                    self.release_workers()
                    yield self.env.process(self.set_to_waiting())

                self.state['processing_time'].update(total_processing_time)

                # Wait to place carrier to buffer_out
                yield self.env.process(self.buffer_out(carrier))
                self.state['carrier'].update(None)

            else:
                yield self.turn_off()

Sink

Bases: Station

The Sink takes carriers from buffer_in. It removes the parts of the carrier and either destroys it or puts them to buffer_out if one exists.

Parameters:

Name	Type	Description	Default
processing_std	float	The standard deviation of the processing time.	None
position	tuple	X and Y position in the visualization.	None

Source code in lineflow/simulation/stations.py

class Sink(Station):
    """
    The Sink takes carriers from buffer_in. It removes the parts of the carrier and either
    destroys it or puts them to buffer_out if one exists.

    Args:
        processing_std (float): The standard deviation of the processing time.
        position (tuple): X and Y position in the visualization.
    """
    def __init__(
        self,
        name,
        buffer_in=None,
        buffer_out=None,
        processing_time=2,
        processing_std=None,
        position=None,
    ):
        super().__init__(
            name=name,
            processing_time=processing_time,
            processing_std=processing_std,
            position=position,
        )

        if buffer_in is not None:
            self._connect_to_input(buffer_in)
        if buffer_out is not None:
            self._connect_to_output(buffer_out)

    def init_state(self):

        self.state = ObjectStates(
            DiscreteState('on', categories=[True, False], is_actionable=False, is_observable=False),
            DiscreteState('mode', categories=['working', 'waiting', 'failing']),
            CountState('n_parts_produced', is_actionable=False, is_observable=False),
            TokenState(name='carrier', is_observable=False),
        )

        self.state['on'].update(True)
        self.state['mode'].update("waiting")
        self.state['n_parts_produced'].update(0)
        self.state['mode'].update("waiting")
        self.state['carrier'].update(None)

    def remove(self, carrier):

        processing_time = self._sample_exp_time(
            time=self.processing_time,
            scale=self.processing_std,
        )
        yield self.env.timeout(processing_time)
        self.state['n_parts_produced'].increment()

        if hasattr(self, 'buffer_out'):
            yield self.env.process(self.set_to_waiting())
            carrier.parts.clear()
            yield self.env.process(self.buffer_out(carrier))

    def run(self):

        while True:
            if self.is_on():
                yield self.env.process(self.set_to_waiting())
                carrier = yield self.env.process(self.buffer_in())
                yield self.env.process(self.set_to_work())
                self.state['carrier'].update(carrier.name)

                # Wait to place carrier to buffer_out
                yield self.env.process(self.remove(carrier))
                self.state['carrier'].update(None)

            else:
                yield self.turn_off()

Source

Bases: Station

Source station generating parts on carriers.

The Source takes carriers from buffer_in, creates a part, places that part onto the carrier, and pushes the carrier onto the buffer_out. If unlimited carriers is True, no buffer_in is needed and no magazine.

Parameters:

Name	Type	Description	Default
name	str	Name of the Cell	required
carrier_specs	dict	Nested dict. Top level descripes carrier types, each consists of a dict specifying different parts setup on the carrier at the source. The part level specifies how the part behaves at future processes along the layout. For instance a spec {'C': {'Part1': {'Process1': {'assembly_condition': 5}, 'Process2': {'extra_processing_time': 10}}}} tells that the produced carrier has one part Part1 that has to fullfill an assembly condition of 5 at station Process1 and gets an additional processing time of 10 at Process2.	None
buffer_in	Buffer	Buffer in	None
buffer_out	obj	Buffer out	None
processing_time	float	Time it takes to put part on carrier (carrier needs to be available)	2
processing_std	float	Standard deviation of processing time	None
waiting_time	float	Time to wait between pushing a carrier out and taking the next one	0
position	tuple	X and Y position in visualization	None
unlimited_carriers	bool	If source has the ability to create unlimited carriers	False
carrier_capacity	int	Defines how many parts can be assembled on a carrier. If set to default (infinity) or > 15, carrier will be visualized with one part.	inf
nok_probability	float	Between 0 and 1, probability of a part on the carrier being labled as nok	0.0
carrier_min_creation	int	Minimum number of carriers of same spec created subsequentially	1
carrier_max_creation	int	Maximum number of carriers of same spec created subsequentially	None

Source code in lineflow/simulation/stations.py

class Source(Station):
    '''
    Source station generating parts on carriers.

    The Source takes carriers from buffer_in, creates a part, places that part
    onto the carrier, and pushes the carrier onto the buffer_out.
    If unlimited carriers is True, no buffer_in is needed and no magazine.

    Args:
        name (str): Name of the Cell
        carrier_specs (dict): Nested dict. Top level descripes carrier types, each consists of a
            dict specifying different parts setup on the carrier at the source. The part level
            specifies how the part behaves at future processes along the layout. For instance a spec
            `{'C': {'Part1': {'Process1': {'assembly_condition': 5}, 'Process2': {'extra_processing_time': 10}}}}` 
            tells that the produced carrier has one part `Part1` that has to fullfill an assembly condition of `5` 
            at station `Process1` and gets an additional processing time of `10` at `Process2`.
        buffer_in (lineflow.simulation.connectors.Buffer, optional): Buffer in
        buffer_out (obj): Buffer out
        processing_time (float): Time it takes to put part on carrier (carrier needs to be
            available)
        processing_std (float): Standard deviation of processing time
        waiting_time (float): Time to wait between pushing a carrier out and taking the next one
        position (tuple): X and Y position in visualization
        unlimited_carriers (bool): If source has the ability to create unlimited carriers
        carrier_capacity (int): Defines how many parts can be assembled on a carrier. If set to
            default (infinity) or > 15, carrier will be visualized with one part.
        nok_probability (float): Between 0 and 1, probability of a part on the carrier being labled
            as nok
        carrier_min_creation (int): Minimum number of carriers of same spec created subsequentially
        carrier_max_creation (int): Maximum number of carriers of same spec created subsequentially

    '''
    def __init__(
        self,
        name,
        buffer_in=None,
        buffer_out=None,
        processing_time=2,
        processing_std=None,
        waiting_time=0,
        waiting_time_step=0.5,
        position=None,
        actionable_magazin=True,
        actionable_waiting_time=True,
        unlimited_carriers=False,
        carrier_capacity=np.inf,
        carrier_specs=None,
        carrier_min_creation=1,
        nok_probability=0.0,
        carrier_max_creation=None,
    ):
        super().__init__(
            name=name,
            position=position,
            processing_time=processing_time,
            processing_std=processing_std,
        )
        self._assert_init_args(unlimited_carriers, carrier_capacity, buffer_in)

        if buffer_in is not None:
            self._connect_to_input(buffer_in)
        if buffer_out is not None:
            self._connect_to_output(buffer_out)

        self.buffer_in_object = buffer_in
        self.unlimited_carriers = unlimited_carriers
        self.carrier_capacity = carrier_capacity
        self.waiting_time_step = waiting_time_step

        self.actionable_magazin = actionable_magazin
        self.actionable_waiting_time = actionable_waiting_time

        if carrier_specs is None:
            carrier_specs = {"carrier": {"part": {}}}
        self.carrier_specs = carrier_specs

        self.unlimited_carriers = unlimited_carriers
        self.carrier_capacity = carrier_capacity
        self.carrier_id = 1
        self.carrier_min_creation = carrier_min_creation
        self.carrier_max_creation = carrier_max_creation if carrier_max_creation is not None else 2*carrier_min_creation

        self._carrier_counter = 0

        self.init_waiting_time = waiting_time
        self.nok_probability = nok_probability
        assert 0 <= self.nok_probability <= 1, "nok_probability must be between 0 and 1"

    def init_state(self):

        self.state = ObjectStates(
            DiscreteState('on', categories=[True, False], is_actionable=False, is_observable=False),
            DiscreteState('mode', categories=['working', 'waiting', 'failing']),
            DiscreteState(
                name='waiting_time', 
                categories=np.arange(0, 100, self.waiting_time_step), 
                is_actionable=self.actionable_waiting_time,
            ),
            TokenState(name='carrier', is_observable=False),
            TokenState(name='part', is_observable=False),
            DiscreteState(
                name='carrier_spec', 
                categories=list(self.carrier_specs.keys()), 
                is_actionable=False,
                is_observable=True,
            ),
        )

        self.state['waiting_time'].update(self.init_waiting_time)
        self.state['on'].update(True)
        self.state['mode'].update("waiting")
        self.state['carrier'].update(None)
        self.state['part'].update(None)
        self.state['carrier_spec'].update(list(self.carrier_specs.keys())[0])

    def _assert_init_args(self, unlimited_carriers, carrier_capacity, buffer_in):
        if unlimited_carriers:
            if carrier_capacity > 15:
                warnings.warn(
                    "If carrier_capacity > 15, visualization of parts"
                    "on carriers is restriced and carrier will be visualized with one part")
            if not isinstance(carrier_capacity, int) and carrier_capacity != np.inf:
                raise AttributeError("Type of carrier capacity must be int or np.inf")

    def create_carrier(self):

        carrier_spec = self.state['carrier_spec'].to_str()

        name = f'{self.name}_{carrier_spec}_{self.carrier_id}'
        carrier = Carrier(
            self.env, 
            name=name, 
            capacity=self.carrier_capacity, 
        )
        self.carrier_id += 1

        return carrier

    def get_current_carrier_spec(self):
        if self._carrier_counter == 0:
            carrier_spec = self.random.choice(list(self.carrier_specs.keys()))
            self.state['carrier_spec'].update(carrier_spec)
            self._carrier_counter = self.random.randint(
                self.carrier_min_creation, 
                self.carrier_max_creation + 1,
            )
        self.carrier_id += 1
        self._carrier_counter -= 1

        return self.carrier_specs[self.state['carrier_spec'].to_str()].copy()


    def create_parts(self, carrier, carrier_spec):
        """
        Creates the parts based on the part_specs attribute
        For each dict in the part_specs list one part is created
        """

        parts = []
        for part_id, (part_name, part_spec) in enumerate(carrier_spec.items()):
            part = Part(
                env=self.env,
                name=f"{carrier.name}_{part_name}_{part_id}",
                specs=part_spec,
                nok_probability=self.nok_probability,
            )
            part.create(self.position)
            parts.append(part)

        return parts

    def assemble_parts_on_carrier(self, carrier, parts):
        """
        Put parts onto carrier
        """
        for part in parts:
            carrier.assemble(part)

    def assemble_carrier(self, carrier, carrier_spec):

        parts = self.create_parts(carrier, carrier_spec)
        self.state['part'].update(parts[0].name)

        processing_time = self._sample_exp_time(
            time=self.processing_time,
            scale=self.processing_std,
        )
        self.state['carrier'].update(carrier.name)

        yield self.env.timeout(processing_time)
        self.assemble_parts_on_carrier(carrier, parts)

        return carrier

    def wait(self):

        waiting_time = self.state['waiting_time'].to_str()

        if waiting_time > 0:
            yield self.env.process(self.set_to_waiting())
            yield self.env.timeout(waiting_time)

    def run(self):

        while True:
            if self.is_on():
                yield self.env.process(self.set_to_waiting())
                yield self.env.process(self.wait())

                if self.unlimited_carriers:
                    carrier = self.create_carrier()
                else:
                    carrier = yield self.env.process(self.buffer_in())

                carrier_spec = self.get_current_carrier_spec()

                yield self.env.process(self.set_to_work())
                carrier = yield self.env.process(self.assemble_carrier(carrier, carrier_spec))

                yield self.env.process(self.set_to_waiting())
                yield self.env.process(self.buffer_out(carrier))
                self.state['part'].update(None)
                self.state['carrier'].update(None)

            else:
                yield self.turn_off()

assemble_parts_on_carrier(carrier, parts)

Put parts onto carrier

Source code in lineflow/simulation/stations.py

def assemble_parts_on_carrier(self, carrier, parts):
    """
    Put parts onto carrier
    """
    for part in parts:
        carrier.assemble(part)

create_parts(carrier, carrier_spec)

Creates the parts based on the part_specs attribute For each dict in the part_specs list one part is created

Source code in lineflow/simulation/stations.py

def create_parts(self, carrier, carrier_spec):
    """
    Creates the parts based on the part_specs attribute
    For each dict in the part_specs list one part is created
    """

    parts = []
    for part_id, (part_name, part_spec) in enumerate(carrier_spec.items()):
        part = Part(
            env=self.env,
            name=f"{carrier.name}_{part_name}_{part_id}",
            specs=part_spec,
            nok_probability=self.nok_probability,
        )
        part.create(self.position)
        parts.append(part)

    return parts

Station

Bases: StationaryObject

Source code in lineflow/simulation/stations.py

class Station(StationaryObject):

    _width = 30
    _height = 30
    _color = 'black'

    def __init__(
        self,
        name,
        position=None,
        processing_time=5,
        processing_std=None,
        rework_probability=0,
        worker_pool=None,
    ):

        super().__init__()

        if position is None:
            position = (0, 0)

        self.name = name
        self.position = pygame.Vector2(position[0], position[1])

        self.worker_pool = worker_pool
        self.worker_requests = {}

        if self.worker_pool is not None:
            self.worker_pool.register_station(self)

        self.processing_time = processing_time
        self.rework_probability = rework_probability

        if self.rework_probability > 1 or self.rework_probability < 0:
            raise ValueError('rework_probability should should be between 0 and 1')

        if processing_std is None:

            self.processing_std = 0.1*self.processing_time
        else:
            assert processing_std >= 0 and processing_std <= 1
            self.processing_std = processing_std*self.processing_time

        self.worker_assignments = {}

    @property
    def is_automatic(self):
        return self.worker_pool is None

    @property
    def n_workers(self):
        if self.worker_pool is not None:
            return len(self.worker_assignments) + 1
        else:
            return 1

    @property
    def worker_skill(self):
        if self.worker_pool is not None:
            skills = [worker.skill_levels.get(self.name, 1.0) for worker in self.worker_assignments.values()]
            skill_sum = np.sum(skills) + 1
            return skill_sum
        else:
            return 1.0

    def setup_draw(self):

        self._rect = pygame.Rect(
            self.position.x - self._width / 2,
            self.position.y - self._height / 2,
            self._width,
            self._height,
        )

        font = pygame.font.SysFont(None, 20)
        self._text = font.render(self.name, True, 'black')

    def _draw(self, screen):
        pygame.draw.rect(screen, self._color, self._rect, border_radius=5)
        self._draw_info(screen)
        screen.blit(
            self._text,
            self._text.get_rect(center=self.position + (0, -0.6 * self._height)),
        )

    def _draw_info(self, screen):
        pass

    def _is_nok_part(self, component):
        return self.random.choice(
            [True, False], 
            p=[component.nok_probability, 1 - component.nok_probability],
        )

    def _draw_n_workers(self, screen):
        if not self.is_automatic:
            font = pygame.font.SysFont(None, 14)
            text = font.render(
                "W=" + str(self.worker_skill),
                True,
                'black',
            )
            screen.blit(
                text,
                text.get_rect(center=self.position),
            )

    def _draw_n_carriers(self, screen):
        font = pygame.font.SysFont(None, 14)
        text = font.render(
            "C=" + self.state['carriers_in_magazine'].to_str(),
            True,
            'black',
        )
        screen.blit(
            text,
            text.get_rect(center=self.position),
        )

    def get_performance_coefficient(self):
        return compute_performance_coefficient(self.worker_skill)

    def _sample_exp_time(self, time=None, scale=None, rework_probability=0):
        """
        Samples a time from an exponential distribution
        """

        coeff = self.get_performance_coefficient()
        t = time * coeff + self.random.exponential(scale=scale)

        rework = self.random.choice(
            [1, 2],
            p=[1-rework_probability, rework_probability],
        )

        return t*rework

    def set_to_waiting(self):
        yield self.env.timeout(0)
        self._color = 'yellow'
        self.state['mode'].update('waiting')
        yield self.env.timeout(0)

    def request_workers(self):
        """
        Requests (and blocks) the worker for the process coming up.
        """
        if not self.is_automatic:
            self.worker_assignments = self.worker_pool.get_worker(self.name)

            self.worker_requests = {
                name: worker.request() for name, worker in self.worker_assignments.items()
            }

            # Block workers for this process
            for request in self.worker_requests.values():
                yield request

        else:
            yield self.env.timeout(0)

    def release_workers(self):
        """
        Releases the worker, to they may follow a new assignment
        """
        if not self.is_automatic:

            for worker, request in self.worker_requests.items():
                self.worker_assignments[worker].release(request)
            self.worker_requests = {}
            self.worker_assignments = {}

    def set_to_error(self):
        yield self.env.timeout(0)
        self._color = 'red'
        self.state['mode'].update('failing')
        yield self.env.timeout(0)

    def set_to_work(self):
        yield self.env.timeout(0)
        self._color = 'green'
        self.state['mode'].update('working')
        yield self.env.timeout(0)

    def turn_off(self):
        self._color = 'gray'
        self.state['on'].update(False)
        self.turn_off_event = simpy.Event(self.env)
        return self.turn_off_event

    def is_on(self):
        return self.state['on'].to_str()

    def turn_on(self):
        event = self.turn_off_event

        self.state['on'].update(True)
        if not event.triggered:
            yield event.succeed()
        else:
            yield self.env.timeout(0)

    def connect_to_input(self, station, *args, **kwargs):
        buffer = Buffer(name=f"Buffer_{station.name}_to_{self.name}", *args, **kwargs)
        self._connect_to_input(buffer)
        station._connect_to_output(buffer)

    def connect_to_output(self, station, *args, **kwargs):
        buffer = Buffer(name=f"Buffer_{self.name}_to_{station.name}", *args, **kwargs)
        self._connect_to_output(buffer)
        station._connect_to_input(buffer)

    def _connect_to_input(self, buffer):
        if hasattr(self, 'buffer_in'):
            raise ValueError(f'Input of {self.name} already connected')
        self.buffer_in = buffer.connect_to_output(self)

    def _connect_to_output(self, buffer):
        if hasattr(self, 'buffer_out'):
            raise ValueError(f'Output of {self.name} already connected')
        self.buffer_out = buffer.connect_to_input(self)

    def register(self, env):
        self.env = env
        self.env.process(self.run())

    def _derive_actions_from_new_state(self, state):
        # Turn machine back on if needed
        if not self.is_on() and 'on' in state and hasattr(self, 'turn_off_event') and state['on'] == 0:
            self.env.process(self.turn_on())

    def apply(self, actions):
        self._derive_actions_from_new_state(actions)
        self.state.apply(actions)

release_workers()

Releases the worker, to they may follow a new assignment

Source code in lineflow/simulation/stations.py

def release_workers(self):
    """
    Releases the worker, to they may follow a new assignment
    """
    if not self.is_automatic:

        for worker, request in self.worker_requests.items():
            self.worker_assignments[worker].release(request)
        self.worker_requests = {}
        self.worker_assignments = {}

request_workers()

Requests (and blocks) the worker for the process coming up.

Source code in lineflow/simulation/stations.py

def request_workers(self):
    """
    Requests (and blocks) the worker for the process coming up.
    """
    if not self.is_automatic:
        self.worker_assignments = self.worker_pool.get_worker(self.name)

        self.worker_requests = {
            name: worker.request() for name, worker in self.worker_assignments.items()
        }

        # Block workers for this process
        for request in self.worker_requests.values():
            yield request

    else:
        yield self.env.timeout(0)

Switch

Bases: Station

A Switch distributes carriers onto buffer outs. In and out buffers can be provided to the constructor but can also be added to a switch using the connect_to_input and connect_to_output methods.

Parameters:

Name	Type	Description	Default
buffers_in	list	A list of buffers that lead into the Switch.	None
buffers_out	list	A list of buffers that lead away from the Switch.	None
position	tuple	X and Y position in the visualization.	None
alternate	bool	If True, the Switch switches between the buffers_out; else, only one buffer_out is used.	False

Source code in lineflow/simulation/stations.py

class Switch(Station):
    """
    A Switch distributes carriers onto buffer outs. In and out buffers can be provided to
    the constructor but can also be added to a switch using the `connect_to_input` and `connect_to_output`
    methods.

    Args:
        buffers_in (list): A list of buffers that lead into the Switch.
        buffers_out (list): A list of buffers that lead away from the Switch.
        position (tuple): X and Y position in the visualization.
        alternate (bool): If True, the Switch switches between the buffers_out; else, only one buffer_out is used.
    """

    def __init__(
        self,
        name,
        buffers_in=None,
        buffers_out=None,
        position=None,
        processing_time=5,
        alternate=False,
    ):
        super().__init__(
            name=name,
            position=position,
            processing_time=processing_time,
            # We assume switches do not have variation here
            processing_std=0,
        )

        # time it takes for a model to change buffer_in or buffer_out
        self.readjustment_time = 10

        self.buffer_in = []
        self.buffer_out = []

        if buffers_in is not None:
            for buffer in buffers_in:
                self._connect_to_input(buffer)

        if buffers_out is not None:
            for buffer in buffers_out:
                self._connect_to_output(buffer)

        self.alternate = alternate

    def init_state(self):
        self.state = ObjectStates(
            DiscreteState('on', categories=[True, False], is_actionable=False, is_observable=False),
            DiscreteState('mode', categories=['working', 'waiting', 'failing']),
            DiscreteState(
                name='index_buffer_in',
                categories=list(range(self.n_buffers_in)),
                is_actionable=not self.alternate and self.n_buffers_in > 1
            ),
            DiscreteState(
                name='index_buffer_out',
                categories=list(range(self.n_buffers_out)),
                is_actionable=not self.alternate and self.n_buffers_out > 1),
            TokenState(name='carrier', is_observable=False),
        )
        self.state['index_buffer_in'].update(0)
        self.state['index_buffer_out'].update(0)
        self.state['on'].update(True)
        self.state['mode'].update("waiting")
        self.state['carrier'].update(None)

    @property
    def n_buffers_in(self):
        return len(self.buffer_in)

    @property
    def n_buffers_out(self):
        return len(self.buffer_out)

    def _get_buffer_in_position(self):
        return self.buffer_in[
            self.state['index_buffer_in'].value
        ].__self__._positions_slots[-1]

    def _get_buffer_out_position(self):
        return self.buffer_out[
            self.state['index_buffer_out'].value
        ].__self__._positions_slots[0]

    def _draw_info(self, screen):
        pos_buffer_in = self._get_buffer_in_position()
        pos_buffer_out = self._get_buffer_out_position()

        pos_in = pos_buffer_in + 0.5*(self.position - pos_buffer_in)
        pos_out = pos_buffer_out + 0.5*(self.position - pos_buffer_out)

        pygame.draw.circle(screen, 'gray', self.position, 6)
        for pos in [pos_in, pos_out]:
            pygame.draw.line(screen, "gray", self.position, pos, width=5)

    def _connect_to_input(self, buffer):
        self.buffer_in.append(buffer.connect_to_output(self))

    def _connect_to_output(self, buffer):
        self.buffer_out.append(buffer.connect_to_input(self))

    def _alternate_indices(self):
        self.state['index_buffer_in'].set_next()
        self.state['index_buffer_out'].set_next()

    def look_in(self):
        """
        Checks if part at current buffer_in is available
        """
        buffer_in = self.buffer_in[self.state['index_buffer_in'].value].__self__
        while buffer_in.get_fillstate() == 0:
            yield self.env.timeout(1)
            buffer_in = self.buffer_in[self.state['index_buffer_in'].value].__self__
        return buffer_in

    def look_out(self):
        """
        Checks if space at current buffer_out is available
        """
        buffer_out = self.buffer_out[self.state['index_buffer_out'].value].__self__

        while buffer_out.get_fillstate() == 1:
            yield self.env.timeout(1)
            # check if buffer out changed
            buffer_out = self.buffer_out[self.state['index_buffer_out'].value].__self__
        return buffer_out

    def get(self):
        while True:
            yield self.env.process(self.set_to_waiting())
            buffer_in = yield self.env.process(self.look_in())
            self.getting_process = None
            yield self.env.process(self.set_to_work())
            carrier = yield self.env.process(
                buffer_in.get()
            )
            self.state['carrier'].update(carrier.name)
            return carrier

    def put(self, carrier):
        while True:
            yield self.env.process(self.set_to_waiting())
            buffer_out = yield self.env.process(self.look_out())
            yield self.env.process(buffer_out.put(carrier))
            self.state['carrier'].update(None)
            return

    def run(self):
        while True:
            if self.is_on():
                # Get carrier
                carrier = yield self.env.process(self.get())

                # Process
                yield self.env.process(self.set_to_work())
                yield self.env.timeout(self.processing_time)

                yield self.env.process(self.put(carrier))

                if self.alternate:
                    self._alternate_indices()

            else:
                yield self.turn_off()

look_in()

Checks if part at current buffer_in is available

Source code in lineflow/simulation/stations.py

def look_in(self):
    """
    Checks if part at current buffer_in is available
    """
    buffer_in = self.buffer_in[self.state['index_buffer_in'].value].__self__
    while buffer_in.get_fillstate() == 0:
        yield self.env.timeout(1)
        buffer_in = self.buffer_in[self.state['index_buffer_in'].value].__self__
    return buffer_in

look_out()

Checks if space at current buffer_out is available

Source code in lineflow/simulation/stations.py

def look_out(self):
    """
    Checks if space at current buffer_out is available
    """
    buffer_out = self.buffer_out[self.state['index_buffer_out'].value].__self__

    while buffer_out.get_fillstate() == 1:
        yield self.env.timeout(1)
        # check if buffer out changed
        buffer_out = self.buffer_out[self.state['index_buffer_out'].value].__self__
    return buffer_out

WorkerPool

Bases: StationaryObject

Source code in lineflow/simulation/stations.py

class WorkerPool(StationaryObject):

    def __init__(
        self,
        name,
        n_workers=None,
        transition_time=5,
        skill_levels=None,
    ):
        super().__init__()

        assert n_workers is not None, "Workers have to be set"

        self.name = name
        self.n_workers = n_workers
        self.transition_time = transition_time
        self.stations = []
        self._station_names = []
        self._worker_names = [f"W{i}" for i in range(self.n_workers)]
        self.skill_levels = skill_levels or {}
        self.workers = {
            name: Worker(
                name=name,
                transition_time=self.transition_time,
                skill_levels=self.skill_levels.get(name, {})
            ) for name in self._worker_names
        }

    def register_station(self, station):
        self.stations.append(station)
        self._station_names.append(station.name)

    def init_state(self):
        for worker in self.workers.values():
            worker.init_state(self.stations)

        self.state = ObjectStates(
            *[
                worker.state for worker in self.workers.values()
            ]
        )
        # Distribute worker on stations in round robin fashion
        for worker, station in zip_cycle(self.n_workers, self.n_stations):
            self.state[f"W{worker}"].apply(station)

    @property
    def n_stations(self):
        return len(self.stations)

    def register(self, env):
        self.env = env

        for worker in self.workers.values():
            worker.register(env)

        for worker_n, station in zip_cycle(self.n_workers, self.n_stations):

            worker = self.workers[f"W{worker_n}"]
            # Start working
            self.env.process(worker.work())

    def apply(self, actions):
        """
        This should just update the state of the workers
        """
        for worker, station in actions.items():
            worker_obj = self.workers[worker]
            self.env.process(worker_obj.assign(station))

    def get_worker(self, name):
        # gather these workers assigned to these station
        station = self._station_names.index(name)
        requests = {}

        for worker in self.workers.values():
            # If state of worker equals the station, the worker is blocked for exactly this station
            if worker.state.value == station:
                requests[worker.name] = worker
        return requests

apply(actions)

This should just update the state of the workers

Source code in lineflow/simulation/stations.py

def apply(self, actions):
    """
    This should just update the state of the workers
    """
    for worker, station in actions.items():
        worker_obj = self.workers[worker]
        self.env.process(worker_obj.assign(station))

Buffer

Bases: Connector

Element that connects two different types of stations.

Parameters:

Name	Type	Description	Default
name	str	Name of the station	required
capacity	int	Number of slots the buffer can hold	2
transition_time	float	Time carriers need to traverse the full buffer	10
put_time	float	Time the buffer needs to hand over a carrier to the next element	1
get_time	float	Time the buffer needs to get a carrier to the previous element	1

Source code in lineflow/simulation/connectors.py

class Buffer(Connector):
    """
    Element that connects two different types of stations.

    Args:
        name (str): Name of the station
        capacity (int): Number of slots the buffer can hold
        transition_time (float): Time carriers need to traverse the full buffer
        put_time (float): Time the buffer needs to hand over a carrier to the next element
        get_time (float): Time the buffer needs to get a carrier to the previous element

    """

    def __init__(self, name, capacity=2, transition_time=10, put_time=1, get_time=1, put_std=None):
        super().__init__()
        self.name = name
        self.transition_time_between_slots = transition_time / (capacity - 1)
        self.transition_time = transition_time
        self.put_time = put_time
        self.get_time = get_time
        self.color = "gray"

        # Small registry of carriers
        self.carriers = {}
        self.capacity = capacity

        if put_std is None:
            self.put_std = 0.1*self.put_time
        else:
            self.put_std = put_std

        self._positions_slots = self.capacity * [None]
        self._positions_arrow = self.capacity * [None]

    def init_state(self):

        self.state = ObjectStates(
            DiscreteState('on', categories=[True, False], is_actionable=False, is_observable=False),
            NumericState(name='fill', vmin=0, vmax=1, is_observable=True, is_actionable=False),
        )

        self.state['on'].update(True)
        self.state['fill'].update(0)

    @property
    def n_carriers(self):
        return len(self.carriers)

    def setup_draw(self):

        vec_direction = np.array(
            [
                self._position_output.x - self._position_input.x,
                self._position_output.y - self._position_input.y
            ]
        ) / (self.capacity + 1)

        if vec_direction[0] == 0 and vec_direction[1] == 0:
            raise ValueError(f'Start and end position of {self.name} equal!')

        vec_start = np.array([self._position_input.x, self._position_input.y])

        for i in range(self.capacity):

            pos = vec_start + (i+1)*vec_direction
            self._positions_slots[i] = pygame.Vector2(pos[0], pos[1])

        # Calculate arrowhead points
        arrowhead_size = 3.5
        vec_normalized = vec_direction / np.linalg.norm(vec_direction)

        for i in range(self.capacity):
            positions_arrow = vec_start + (i+1.55) * vec_direction
            arrowhead = np.tile(positions_arrow, (3, 1))

            for j, index in enumerate([0, 2]):
                arrowhead[index] += arrowhead_size*np.array(
                    [
                        -vec_normalized[0]+(-1)**j*vec_normalized[1],
                        -vec_normalized[1]-(-1)**j*vec_normalized[0],
                    ]
                )

            self._positions_arrow[i] = arrowhead

    def _draw(self, screen):

        pygame.draw.line(
            screen,
            self.color,
            self._position_input,
            self._position_output,
            width=10,
        )

        # Draw slots
        for i, slot in enumerate(self._positions_slots):
            pygame.draw.circle(screen, 'gray', slot, 10)

        # Draw arrowheads
        for i, arrow in enumerate(self._positions_arrow[:-1]):
            pygame.draw.polygon(screen, 'black', arrow)

        # Draw carriers
        for carrier in self.carriers.values():
            carrier.draw(screen)

    def _sample_put_time(self):
        return self.put_time + self.random.exponential(scale=self.put_std)

    def get(self):
        self.carrier = yield self.slots[-1].get()
        # Release segment
        yield self.blockers[-1].get()

        # Remove carrier from registry
        self.carriers.pop(self.carrier.name)
        self.state['fill'].update(self.get_fillstate())
        yield self.env.timeout(self.get_time)
        return self.carrier

    def put(self, carrier):

        # Wait for segment to be released
        yield self.blockers[0].put(1)

        # Wait a bit to actually put part
        yield self.env.timeout(self._sample_put_time())

        yield self.slots[0].put(carrier)

        carrier.move(self._positions_slots[0])

        # Remove carrier from registry
        self.carriers[carrier.name] = carrier
        self.state['fill'].update(self.get_fillstate())

    def get_fillstate(self):
        return self.n_carriers/self.capacity

    def _move(self, i):

        slot_from = self.slots[i]
        slot_to = self.slots[i+1]

        blocker_self = self.blockers[i]
        blocker_next = self.blockers[i+1]

        while True:
            # Wait for carrier in slot
            carrier = yield slot_from.get()

            # Waiting for segment to be free
            yield blocker_next.put(1)

            carrier.move(self._positions_slots[i+1])

            # Release segment before
            yield blocker_self.get()

            # Start moving Transition
            yield self.env.timeout(self.transition_time_between_slots)

            # Push carrier to next segment
            yield slot_to.put(carrier)

    def register(self, env):
        self.env = env

        self.slots = [
            simpy.Store(
                env=self.env,
                capacity=1,
            ) for _ in range(self.capacity)
        ]

        self.blockers = [
            simpy.Store(
                env=self.env,
                capacity=1,
            ) for _ in range(self.capacity)
        ]

        for i in range(self.capacity - 1):
            self.env.process(self._move(i))

Carrier

Bases: MovableObject

Source code in lineflow/simulation/movable_objects.py

class Carrier(MovableObject):

    def __init__(self, env, name, color='Black', width=30, height=10, capacity=np.inf):
        super(Carrier, self).__init__(env, name, specs=None)
        self.capacity = capacity
        self._color = color
        self._width = width
        self._height = height

        self._width_part = 0.8*self._width
        if capacity < np.inf:
            self._width_part = self._width_part / self.capacity

        self._height_part = 0.7*self._height

        self.parts = {}

    def assemble(self, part):

        if part.name in self.parts:
            raise ValueError(f'A part with name {part.name} already contained')

        if not hasattr(part, "creation_time"):
            raise ValueError('Part not created')

        if self.capacity == len(self.parts):
            raise ValueError('Carrier is already full. Check your carrier_capacity')

        self.parts[part.name] = part

    def _draw_shape(self, screen):

        self._rect = pygame.Rect(
            self._position.x - self._width / 2,
            self._position.y - self._height / 2,
            self._width,
            self._height,
        )
        pygame.draw.rect(screen, self._color, self._rect, border_radius=2)

        for i, part in enumerate(self):
            part._draw(
                screen,
                x=self._position.x+0.1*self._width - self._width / 2 + i*(self._width_part),
                y=self._position.y - self._height_part / 2,
                width=self._width_part,
                height=self._height_part,
            )

    def move(self, position):
        """
        """

        # If no position has been given, no move is taking place
        if position is None:
            return

        if not isinstance(position, pygame.Vector2):
            raise ValueError('Expect pygame vector as position')

        self._position = position

        for part in self.parts.values():
            part.move(position)

    def __iter__(self):
        for part in self.parts.values():
            yield part

    def get_parts_for_station(self, station):
        return [p for p in self if station in p.specs]

    def get_additional_processing_time(self, station):
         total_time = 0

         for part in self:
             processing_time = part.specs.get(station, {}).get("extra_processing_time", 0)
             total_time += processing_time

         return total_time

move(position)

Source code in lineflow/simulation/movable_objects.py

def move(self, position):
    """
    """

    # If no position has been given, no move is taking place
    if position is None:
        return

    if not isinstance(position, pygame.Vector2):
        raise ValueError('Expect pygame vector as position')

    self._position = position

    for part in self.parts.values():
        part.move(position)

Part

Bases: MovableObject

Source code in lineflow/simulation/movable_objects.py

class Part(MovableObject):
    def __init__(self, env, name, specs=None, color='Orange', nok_probability=0.0):
        super(Part, self).__init__(env, name, specs=specs)
        self.nok_probability = nok_probability
        self._color = color

    def is_valid_for_assembly(self, station_name):
        """
        If the part has an `assembly_condition` in its specification, then it checks whether the
        time between its creation and now is smaller than this condition. Otherwise it will just
        return true.
        """
        if "assembly_condition" in self.specs.get(station_name, {}):
            return (self.env.now - self["creation_time"]) < self.specs[station_name]["assembly_condition"]
        else:
            return True

    def create(self, position):
        if not isinstance(position, pygame.Vector2):
            raise ValueError('Expect pygame vector as position')
        self.move(position)

    def _draw(self, screen, x, y, width, height):
        _part_rect = pygame.Rect(x, y, width, height)
        pygame.draw.rect(screen, self._color, _part_rect, border_radius=1)

    def get_processing_time(self, station):
        return self.specs.get(station, {}).get("extra_processing_time", 0)

    def get_error_probability(self, station):
        return self.specs.get(station, {}).get("error_probability", 0.0)

    def get_error_time(self, station):
        return self.specs.get(station, {}).get("error_time", 0)

is_valid_for_assembly(station_name)

If the part has an assembly_condition in its specification, then it checks whether the time between its creation and now is smaller than this condition. Otherwise it will just return true.

Source code in lineflow/simulation/movable_objects.py

def is_valid_for_assembly(self, station_name):
    """
    If the part has an `assembly_condition` in its specification, then it checks whether the
    time between its creation and now is smaller than this condition. Otherwise it will just
    return true.
    """
    if "assembly_condition" in self.specs.get(station_name, {}):
        return (self.env.now - self["creation_time"]) < self.specs[station_name]["assembly_condition"]
    else:
        return True

Worker

Bases: object

Source code in lineflow/simulation/movable_objects.py

class Worker(object):
    def __init__(self, name, transition_time=5, skill_levels=None):
        self.name = name
        self.transition_time = transition_time

        if skill_levels is None:
            skill_levels = {}
        self.skill_levels = skill_levels

    def register(self, env):
        self.env = env
        self._working = Resource(self.env, capacity=1)
        self.assignment = Store(env=self.env, capacity=1)

    def release(self, request):
        self._working.release(request)

    def request(self):
        return self._working.request()

    def get_skill_level(self, station_name):
        return self.skill_levels.get(station_name, 1.0)

    def assign(self, station):
        """
        Assign worker to station.

        """
        if len(self.assignment.items) > 0:
            # Clean old assignment
            yield self.assignment.get()

        yield self.assignment.put(station)

    def init_state(self, stations):

        self.stations = stations
        self.state = DiscreteState(
            name=self.name,
            categories=[s.name for s in self.stations],
            is_observable=True,
            is_actionable=True,
        )

    def work(self):

        # Initially fill value of state to assignment
        yield self.assignment.put(self.state.value)

        while True:
            # Wait for new assignment
            station = yield self.assignment.get()

            # Wait until worker is released from current station
            transition_request = self.request()
            yield transition_request

            if self.state.value != station:
                # New cell-assignment, wait for transition
                # Move to new cell
                yield self.env.timeout(self.transition_time)

            self.release(transition_request)
            # Station now can create requests
            self.state.apply(station)

assign(station)

Assign worker to station.

Source code in lineflow/simulation/movable_objects.py

def assign(self, station):
    """
    Assign worker to station.

    """
    if len(self.assignment.items) > 0:
        # Clean old assignment
        yield self.assignment.get()

    yield self.assignment.put(station)

StationaryObject

Source code in lineflow/simulation/stationary_objects.py

class StationaryObject:
    _objects = []

    def __init__(self):
        StationaryObject._objects.append(self)

    def setup_draw(self):
        pass

    def _draw(self, screen):
        pass

    def register(self, env):
        raise NotImplementedError

    def __enter__(self):
        # Clean up line objects
        StationaryObject._objects = []

        return self._objects

    def init(self, random):
        """
        Function that is called after line is built, so all available information is present
        """
        self.random = random
        self.init_state()

    def init_state(self):
        """
        Should initialize the state object
        """
        raise NotImplementedError()

    def __exit__(self, type, value, traceback):
        StationaryObject._objects = []

init(random)

Function that is called after line is built, so all available information is present

Source code in lineflow/simulation/stationary_objects.py

def init(self, random):
    """
    Function that is called after line is built, so all available information is present
    """
    self.random = random
    self.init_state()

init_state()

Should initialize the state object

Source code in lineflow/simulation/stationary_objects.py

def init_state(self):
    """
    Should initialize the state object
    """
    raise NotImplementedError()

This file is a wrapper around our simulation such that it is compatible with the stable-baselines repo

LineSimulation

Bases: Env

A Gym-compatible environment for simulating production lines using LineFlow.

This environment wraps a LineFlow production line into a reinforcement learning setup, where agents can interact with stations via actions at discrete decision points, while the underlying process unfolds in continuous time using discrete-event simulation.

Parameters:

Name	Type	Description	Default
line	Line	A LineFlow Line object representing the production line layout and behavior.	required
simulation_end	int	The simulation end time (in simulation time units).	required
reward	str	The reward signal to use. Options are "parts" (default) for counting produced parts, or "uptime" for average utilization.	'parts'
part_reward_lookback	int	Time window for computing average uptime-based rewards (used only if reward=uptime).	0
render_mode	str or None	Optional rendering mode. Currently supports "human" for visual rendering or None.	None

Source code in lineflow/simulation/environment.py

class LineSimulation(gym.Env):
    """
    A Gym-compatible environment for simulating production lines using LineFlow.

    This environment wraps a LineFlow production line into a reinforcement learning setup, where
    agents can interact with stations via actions at discrete decision points, while the underlying
    process unfolds in continuous time using discrete-event simulation.

    Args:
        line (lineflow.simulation.line.Line): A LineFlow `Line` object representing the production
            line layout and behavior.
        simulation_end (int): The simulation end time (in simulation time units).
        reward (str, optional): The reward signal to use. Options are "parts" (default) for counting
            produced parts, or "uptime" for average utilization.
        part_reward_lookback (int, optional): Time window for computing average uptime-based rewards
            (used only if reward=`uptime`).
        render_mode (str or None, optional): Optional rendering mode. Currently supports "human" for
            visual rendering or None.
    """

    metadata = {"render_modes": [None, "human"]}

    def __init__(self, line, simulation_end, reward="parts", part_reward_lookback=0, render_mode=None):
        super().__init__()
        self.line = line
        self.simulation_end = simulation_end
        self.part_reward_lookback = part_reward_lookback

        self.render_mode = render_mode

        assert reward in ["uptime", "parts", "parts-sparse"]
        self.reward = reward

        # fix an order of states
        self.action_space = _build_action_space(self.line.state)
        self.observation_space = _build_observation_space(line_state=self.line.state)

        self.n_parts = 0
        self.n_scrap_parts = 0

    def _map_to_action_dict(self, actions):

        actions_iterator = filterfalse(
            lambda n: not self.line.state[n[0]][n[1]].is_actionable,
            self.line.state
        )

        actions_dict = {}
        for action, (station, action_name) in zip(actions, actions_iterator):
            if station not in actions_dict:
                actions_dict[station] = {}

            actions_dict[station][action_name] = action
        return actions_dict

    def step(self, actions):
        """
        Advances the simulation by one environment step.

        Args:
            actions (list or array): A list of agent actions corresponding to actionable features.

        Returns:
            observation (numpy.ndarray): Observation tensor representing the current state.
            reward (float): The computed reward for the current step.
            terminated (bool): Whether the episode has ended.
            truncated (bool): Whether the episode ended due to an internal error or simulation limit.
            info (dict): Additional diagnostic information.
        """
        actions = self._map_to_action_dict(actions)
        self.line.apply(actions)

        try:
            state, terminated = self.line.step(self.simulation_end)
            truncated = False
        except simpy.core.EmptySchedule:
            # TODO: not tested yet
            state = self.line.state
            terminated = True
            truncated = True

        observation = self._get_observations_as_tensor(state)

        if self.reward == "parts":
            reward = (self.line.get_n_parts_produced() - self.n_parts) - \
                self.line.scrap_factor*(self.line.get_n_scrap_parts() - self.n_scrap_parts)
        elif self.reward == "uptime":
            reward = self.line.get_uptime(lookback=self.part_reward_lookback).mean()
        elif self.reward == "parts-sparse":
            reward = 0
            if terminated:
                reward = self.line.get_n_parts_produced()-self.line.scrap_factor*self.line.get_n_scrap_parts()
        else:
            raise ValueError(f"Reward {self.reward} not implemented")

        self.n_parts = self.line.get_n_parts_produced()
        self.n_scrap_parts = self.line.get_n_scrap_parts()

        if self.render_mode == "human":
            self.render()

        return observation, reward, terminated, truncated, self._get_info()

    def _get_info(self):
        return self.line.info()

    def increase_scrap_factor(self, factor=0.1):
        """
        Sets the scrap penalty factor in the reward function.

        Args:
            factor (float): The multiplier applied to scrap parts in the parts-based reward.
        """
        self.line.scrap_factor = factor

    def reset(self, seed=None, options=None):
        gym.Env.reset(self, seed=seed)

        self.line.reset(random_state=seed)
        self.n_parts = 0
        self.n_scrap_parts = 0

        state, _ = self.line.step()
        observation = self._get_observations_as_tensor(state)

        if self.render_mode == "human":
            self.line.setup_draw()
            self.render()
        return observation, self._get_info()

    @property
    def features(self):
        return self.line.state.observable_features

    def render(self):
        self.line._draw()

    def close(self):
        if self.render_mode == 'human':
            self.line.viewpoint.teardown()

    def _get_observations_as_tensor(self, state):

        X = state.get_observations(lookback=1, include_time=False)
        return np.array(X, dtype=np.float32)

increase_scrap_factor(factor=0.1)

Sets the scrap penalty factor in the reward function.

Parameters:

Name	Type	Description	Default
factor	float	The multiplier applied to scrap parts in the parts-based reward.	0.1

Source code in lineflow/simulation/environment.py

def increase_scrap_factor(self, factor=0.1):
    """
    Sets the scrap penalty factor in the reward function.

    Args:
        factor (float): The multiplier applied to scrap parts in the parts-based reward.
    """
    self.line.scrap_factor = factor

step(actions)

Advances the simulation by one environment step.

Parameters:

Name	Type	Description	Default
actions	list or array	A list of agent actions corresponding to actionable features.	required

Returns:

Name	Type	Description
observation	ndarray	Observation tensor representing the current state.
reward	float	The computed reward for the current step.
terminated	bool	Whether the episode has ended.
truncated	bool	Whether the episode ended due to an internal error or simulation limit.
info	dict	Additional diagnostic information.

Source code in lineflow/simulation/environment.py

def step(self, actions):
    """
    Advances the simulation by one environment step.

    Args:
        actions (list or array): A list of agent actions corresponding to actionable features.

    Returns:
        observation (numpy.ndarray): Observation tensor representing the current state.
        reward (float): The computed reward for the current step.
        terminated (bool): Whether the episode has ended.
        truncated (bool): Whether the episode ended due to an internal error or simulation limit.
        info (dict): Additional diagnostic information.
    """
    actions = self._map_to_action_dict(actions)
    self.line.apply(actions)

    try:
        state, terminated = self.line.step(self.simulation_end)
        truncated = False
    except simpy.core.EmptySchedule:
        # TODO: not tested yet
        state = self.line.state
        terminated = True
        truncated = True

    observation = self._get_observations_as_tensor(state)

    if self.reward == "parts":
        reward = (self.line.get_n_parts_produced() - self.n_parts) - \
            self.line.scrap_factor*(self.line.get_n_scrap_parts() - self.n_scrap_parts)
    elif self.reward == "uptime":
        reward = self.line.get_uptime(lookback=self.part_reward_lookback).mean()
    elif self.reward == "parts-sparse":
        reward = 0
        if terminated:
            reward = self.line.get_n_parts_produced()-self.line.scrap_factor*self.line.get_n_scrap_parts()
    else:
        raise ValueError(f"Reward {self.reward} not implemented")

    self.n_parts = self.line.get_n_parts_produced()
    self.n_scrap_parts = self.line.get_n_scrap_parts()

    if self.render_mode == "human":
        self.render()

    return observation, reward, terminated, truncated, self._get_info()

Reinforcement Learning

train(config)

Function that handles RL training

Args: - train: Scores from the model update phase - rollout: Scores when a policy is rolled out to gather new experiences. - eval: Scores when a policy is evaluated on a separate environment

Notes

Size of rollout-buffer: n_steps*n_envs, then an model-update is done

Source code in scripts/train.py

def train(config):
    """
    Function that handles RL training

    Args:
    - `train`: Scores from the model update phase
    - `rollout`: Scores when a policy is rolled out to gather new experiences.
    - `eval`: Scores when a policy is evaluated on a separate environment

    Notes:
        Size of rollout-buffer: `n_steps*n_envs`, then an model-update is done
    """

    simulation_end = config['simulation_end'] + 1

    env_train = make_stacked_vec_env(
        line=_make_line(config['env'], config['n_cells'], config['info'], curriculum=config['curriculum']),
        simulation_end=simulation_end,
        reward=config["rollout_reward"],
        n_envs=config['n_envs'],
        n_stack=config['n_stack'] if not config['recurrent'] else 1,
    )

    env_eval = make_stacked_vec_env(
        line=_make_line(config['env'], config['n_cells'], config['info'], curriculum=config['curriculum']),
        simulation_end=simulation_end,
        reward=config["eval_reward"],
        n_envs=1,
        n_stack=config['n_stack'] if not config['recurrent'] else 1,
    )
    run = wandb.init(
        project='Lineflow',
        sync_tensorboard=True,
        config=config
    )
    log_path = os.path.join(config['log_dir'], run.id)

    if config['env'] == 'complex_line' and config['curriculum']:
        curriculum_callback = CurriculumLearningCallback(
            # Task marked as resolved if rewards is above 100
            threshold=100, 
            # Update of scrap factor
            update=(1/config["n_cells"])/5, 
            factor_max=1/config["n_cells"],
            look_back=3,
        )
    else:
        curriculum_callback = None

    eval_callback = EvalCallback(
        eval_env=env_eval,
        deterministic=config['deterministic'],
        n_eval_episodes=1,
        # Every (eval_freq*eval_envs) / (n_steps*train_envs)  step an update is done
        eval_freq=config["n_steps"]* config["n_envs"] * 10, # ever step in every env counts
        callback_after_eval=curriculum_callback,
    )

    model_args = {
        "policy": 'MlpPolicy' if not config['recurrent'] else 'MlpLstmPolicy',
        "env": env_train,
        "n_steps": config["n_steps"],
        "gamma": config['gamma'],  # discount factor
        "learning_rate": config["learning_rate"],
        "use_sde": False,
        "normalize_advantage": config['normalize_advantage'],
        "device": get_device(),
        "tensorboard_log": log_path,
        "stats_window_size": 10,
        "verbose": 0,
        "seed": config['seed'] if config['seed'] != 0 else None,
    }

    if "PPO" in config['model']:
        model_cls = PPO
        model_args["batch_size"] = config["n_steps"]  # mini-batch size
        model_args["n_epochs"] = 5  # number of times to go over experiences with mini-batches
        model_args["clip_range"] = config['clip_range']
        model_args["max_grad_norm"] = 0.5
        model_args["ent_coef"] = config['ent_coef']

        if config['recurrent']:
            model_cls = RecurrentPPO

    if "A2C" in config['model']:
        model_cls = A2C
        model_args["max_grad_norm"] = 0.5
        model_args["ent_coef"] = config['ent_coef']

    if "TRPO" in config['model']:
        model_cls = TRPO

    model = model_cls(**model_args)

    model.learn(
        total_timesteps=config["total_steps"],
        callback=CallbackList([
            WandbCallback(verbose=0),
            eval_callback,
        ])
    )
    run.finish()

CurriculumLearningCallback

Bases: BaseCallback

Parameters:

Name	Type	Description	Default
look_back	int	Number of last evaluations that should be used to check if task is solved	5
factor_max	float	Factor with which the threshold is multiplied to get new threshold	inf

Source code in lineflow/learning/curriculum.py

class CurriculumLearningCallback(BaseCallback):
    """

    Args:
        look_back (int): Number of last evaluations that should be used to check if task is solved
        factor_max (float): Factor with which the threshold is multiplied to get new threshold
    """
    parent: EvalCallback

    def __init__(self, threshold, update=0.02, factor_max=np.inf, look_back=5):
        super().__init__(verbose=0)

        self.threshold = threshold
        self.look_back = look_back
        self.update = update
        self.factor = 0
        self.factor_max = factor_max

        self.rewards = []
        self.last_adjustment = 0

    def update_task(self):
        self.factor = min(self.factor + self.update, self.factor_max)
        self.parent.eval_env.env_method('increase_scrap_factor', (self.factor))
        self.parent.training_env.env_method('increase_scrap_factor', (self.factor))

    def _on_step(self) -> bool:
        self.rewards.append(self.parent.last_mean_reward)
        if self.n_calls >= self.last_adjustment + self.look_back:

            # Check if task is solved
            if np.mean(self.rewards[-self.look_back:]) >= self.threshold:
                self.update_task()
                self.last_adjustment = self.n_calls
        self.logger.record("eval/scrap_factor", self.factor)
        return True

Examples

PartDependentProcessLine

Bases: Line

Example implementation of a line with part-dependent processing times. The source creates carriers of two types (Type_A and Type_B), each containing two parts (Part1 and Part2). The processing times for each process depend on the type of carrier and the parts it contains. The line consists of three processes (P1, P2, P3) and a sink. The processing times for each process are defined in the carrier_specs of the source, where extra processing time is added based on the type of carrier and the parts it contains.

Source code in lineflow/examples/part_dependence.py

class PartDependentProcessLine(Line):
    """
    Example implementation of a line with part-dependent processing times. The source creates
    carriers of two types (Type_A and Type_B), each containing two parts (Part1 and Part2). The
    processing times for each process depend on the type of carrier and the parts it contains. The
    line consists of three processes (P1, P2, P3) and a sink. The processing times for each process
    are defined in the carrier_specs of the source, where extra processing time is added based on
    the type of carrier and the parts it contains.
    """
    def build(self):
        source = Source(
            name='Source',
            processing_time=10,
            position=(100, 200),
            unlimited_carriers=True,
            carrier_capacity=2,
            carrier_min_creation=10,
            carrier_specs={
                'Type_A': {
                    'Part1': {
                        'P1': {'extra_processing_time': 20}, 
                        'P2': {'extra_processing_time': 0},
                        'P3': {'extra_processing_time': 0},
                    },
                    'Part2': {
                        'P1': {'extra_processing_time': 20}, 
                        'P2': {'extra_processing_time': 0},
                        'P3': {'extra_processing_time': 0},
                    }
                },
                'Type_B': {
                    'Part1': {
                        'P1': {'extra_processing_time': 5}, 
                        'P2': {'extra_processing_time': 20},
                        'P3': {'extra_processing_time': 0},
                    },
                    'Part2': {
                        'P1': {'extra_processing_time': 5}, 
                        'P2': {'extra_processing_time': 20},
                        'P3': {'extra_processing_time': 0},
                    }
                },
            },
        )

        p1 = Process('P1', processing_time=10, position=(350, 200))
        p2 = Process('P2', processing_time=10, position=(600, 200))
        p3 = SequentialProcess('P3', processing_time=5, position=(850, 200))
        sink = Sink('Sink', position=(950, 200))

        p1.connect_to_input(source, capacity=10)
        p2.connect_to_input(p1, capacity=10)
        p3.connect_to_input(p2, capacity=10)
        sink.connect_to_input(p3)

WTAssembly

Bases: Assembly

Source code in lineflow/examples/waiting_time.py

class WTAssembly(Assembly):

    def __init__(
        self,
        name,
        R=0.75,
        t_jump_max=2000,
        **kwargs,
    ):

        self.R = R
        self.t_jump_max = t_jump_max
        self.trigger_time = None
        self.factor = None

        super().__init__(name=name,  **kwargs)

    def init(self, random):
        """
        Function that is called after line is built, so all available information is present
        """
        super().init(random)

        self._sample_trigger_time()

    def _compute_scaling_factor(self, T_jump, E=3.1):

        T = self.processing_time
        S = self.processing_std
        T_sim = self.t_jump_max*2

        return 1/T*((T_jump*(T+S+E)) / ((self.R-1)*T_sim+T_jump) - S -E)


    def _sample_trigger_time(self):

        self.t_jump = np.random.uniform(
            0.8*self.t_jump_max,
            self.t_jump_max,
        )

        self.factor = self._compute_scaling_factor(self.t_jump)
        self.trigger_time = self.random.uniform(0.25, 0.75)*self.t_jump_max

    def _sample_exp_time(self, time=None, scale=None, rework_probability=0):
        """
        Samples a time from an exponential distribution
        """
        coeff = self.get_performance_coefficient()
        if self.trigger_time < self.env.now < self.trigger_time + self.t_jump:
            factor = self.factor
        else: 
            factor = 1

        return time*factor*coeff + self.random.exponential(scale=scale)

init(random)

Function that is called after line is built, so all available information is present

Source code in lineflow/examples/waiting_time.py

def init(self, random):
    """
    Function that is called after line is built, so all available information is present
    """
    super().init(random)

    self._sample_trigger_time()

MultiSink

Bases: Line

Assembly line with two sources a switch and a sink

Source code in lineflow/examples/multi_sink.py

class MultiSink(Line):
    """
    Assembly line with two sources a switch and a sink
    """

    def __init__(self, alternate=True, n_sinks=5, **kwargs):
        self.alternate = alternate
        self.n_sinks = n_sinks
        super().__init__(**kwargs)

    def build(self):

        source = Source(
            name='Source',
            processing_time=5,
            position=(100, 300),
            processing_std=0,
            unlimited_carriers=True,
            carrier_capacity=1,
            actionable_magazin=False,
            actionable_waiting_time=False,
        )

        switch = Switch(
            'Switch',
            position=(300, 300),
            alternate=self.alternate,
            processing_time=0,
        )

        source.connect_to_output(switch, capacity=2, transition_time=5)

        for i in range(self.n_sinks):
            sink = Sink(
                name=f'Sink_{i}',
                position=(500, 500-100*i),
                processing_time=10+10*i*3,
                processing_std=0,
            )
            sink.connect_to_input(switch, capacity=4, transition_time=5)

MultiProcess

Bases: Line

Assembly line with two sources a switch and a sink

Source code in lineflow/examples/multi_process.py

class MultiProcess(Line):
    """
    Assembly line with two sources a switch and a sink
    """

    def __init__(self, alternate=True, n_processes=5, **kwargs):
        self.alternate = alternate
        self.n_processes = n_processes
        super().__init__(**kwargs)

    def build(self):
        source = Source(
            name='Source',
            processing_time=2,
            actionable_magazin=False,
            actionable_waiting_time=False,
            unlimited_carriers=True,
            carrier_capacity=1,
            position=(50, 300),
            processing_std=0,
        )

        switch_d = Switch(
            'SwitchD',
            position=(200, 300),
            alternate=self.alternate,
            processing_time=1,
        )

        switch_f = Switch(
            'SwitchF',
            position=(900, 300),
            alternate=self.alternate,
            processing_time=1,
        )

        processes = []
        for i in range(self.n_processes):
            processes.append(
                Process(
                    name=f'P{i}',
                    processing_time=12+10*i,
                    position=(600, 500-100*i),
                    processing_std=0.1,
                )
            )
        sink = Sink('Sink', position=(1100, 300), processing_time=1, processing_std=0)
        switch_f.connect_to_output(sink, capacity=2, transition_time=5)
        switch_d.connect_to_input(source, capacity=2, transition_time=5)

        for process in processes:
            process.connect_to_input(switch_d, capacity=8, transition_time=7)
            process.connect_to_output(switch_f, capacity=5, transition_time=7)