anybox / powerscan_scenario Goto Github PK

Scenario engine for PowerScan (Datalogic scanner)

Each scenario is described in the class, This class is loaded by the setuptools entry point powerscan.scenario.

AnyBlok is released under the terms of the Mozilla Public License.

This project is open source the package can be installed from:

• pip or a similar tool::

  pip install powerscan_scenario
  

• source on github

  git clone [email protected]:anybox/powerscan_scenario.git
  cd powerscan_scenario
  python setup.py install
  

To run tests with nose:

pip install nose
nosetests powerscan_scenario/tests

The install process will ensure that SQLAlchemy, Alembic, SQLAlchemy-Utils are installed, in addition to other dependencies.

The latest version of them is strongly recommended.

The integrator choose the BBD to use by powerscan

This script load the scenario(s) from the entry points and loop to wait the scan from scanners to play job from scenario

This script load the configuration into a scanner

Each scenario have to inherit from powerscan_scenario.Scenario.

# module.path.myscenario.py

from powerscan_scenario.scenario import Scenario


class MyScenario(Scenario):
    version = '1.0.0'
    label = 'My scenario'
    sequence = 1

//
# setup.py
setup(
    ...
    entry_points={
        'powerscan_scenario.scenario': [
            'myscenario=module.path.myscenario:MyScenario',
        ],
    }
)

These attributes are saved in the table scenario

Some hooks can be overwritten

The decorator powerscan_scenario.decorator.step is a helper to define a step in the scenario

from powerscan_scenario.scenario import Scenario
from powerscan_scenario.decorator import step


class MyScenario(Scenario):
    version = '1.0.0'
    label = 'My scenario'
    sequence = 1

    @step()
    def foo(self, session, scanner, entry):
        # action to do
        return {
            'display': [],  # list of string to display
            'buttons': {},  # button label
            'action_type: '',  # Type of the next action to do
            'sound': '',  # sound to play
        }

These parameters of decorator are saved in the table step

The parameters of step method are

The step method return a dict with some key, this dict over writting their default values

The decorator powerscan_scenario.decorator.transition is a helper to define a transition between steps

from powerscan_scenario.scenario import Scenario
from powerscan_scenario.decorator import step
from powerscan_scenario.decorator import transition


class MyScenario(Scenario):
    version = '1.0.0'
    label = 'My scenario'
    sequence = 1

    @step()
    def foo(self, session, scanner, entry):
        # action to do

    @step()
    def bar(self, session, scanner, entry):
        # action to do

    @transition(froms=['foo'], to='bar', sequence=1)
    def check_transition_from_foo_to_var(self, session, scanner, entry):
        return ...  # True or False

These parameters of decorator are saved in the table transition

The parameters of step method are

the method must return a boolean:

• True: The transition is checked, the step targeting will be executed
• False: pass to the next transition

This model saved the scenario coming from the entry points powerscan_scenario.scenario.

This model saved the step coming from the decorator step.

This model saved the transition coming from the decorator transition.

This model saved the job for one scenario.

This model saved the scanner used for one job. The entries is created by powerscan_scenario.

from powerscan_scenario.scenario import Scenario
from powerscan_scenario.decorator import step
from powerscan_scenario.decorator import transition
from sqlalchemy import Column, String, Integer
from sqlalchemy.orm import relationship
from .api import get_data, send_data


class PutProductInLocation(Scenario):
    version = '1.0.0'
    label = 'Move products'
    sequence = 100
    stop_code = '.....'

    def create_models(self, SQLAbase):

        class ProductLocation(SQLAbase):
            __tablename__ = "product_location"

            job_id = Column(Integer, nullable=False, ForeignKey('job.id')
            job = relationship('Job')
            product = Column(String, nullable=False, primary_key=True)
            location = Column(String, nullable=False)
            location_label = Column(String, nullable=False)
            quantity = Column(Integer)
            quantity_count = Column(Integer, default=0)

        self.ProductLocation = ProductLocation

    def initialize_job(self, session, job):
        for (product, location, location_label, quantity) in get_data():
            session.add(self.ProductLocation(
                job=job, product=product, location=location,
                location_label=location_label, quantity=quantity))

    def release_job(self, session, job):
        query = session.query([self.ProductLocation]).filter_by(job=job)
        send_data([
            (x.product, x.location, x.quantity_count)
            for x in query.filter_by(job=job).all()])

        query.delete()

    @step(is_first_step=True)
    def scan_product(self, session, scanner, entry):
        if entry:
            # come from step scan_location
            query = session.query([self.ProductLocation])
            query = query.filter(self.ProductLocation.job == scanner.job)
            query = query.filter(self.ProductLocation.product == scanner.properties['product'])
            query = query.filter(self.ProductLocation.location == entry)
            query = query.filter(self.ProductLocation.quantity_count < self.ProductLocation.quantity)
            line = query.first()
            line.quantity_count += 1

        scanner.properties = {'location_label': '', 'product': '', location: ''}
        return {
            'display': ['Scan a product'],
        }

    @step(is_first_step=True)
    def scan_another_product(self, session, scanner, entry):
        return {
            'display': ['Scan a product'],
            'sound': self.BadRead,
        }

    @step()
    def scan_location(self, session, scanner, entry):
        sound = self.BadRead
        if not scanner.properties['location_label']:
            query = session.query([self.ProductLocation])
            query = query.filter(self.ProductLocation.job == scanner.job)
            query = query.filter(self.ProductLocation.product == entry)
            query = query.filter(self.ProductLocation.quantity_count < self.ProductLocation.quantity)
            line = query.first()

            scanner.properties.update({'location_label': line.location_label, location: line.location, 'product': entry})
            sound = self.GoodRead

        return {
            'display': ['Scan the location', scanner.properties['location_label']],
            'sound': sound,
        }

    @step()
    def stop(self, session, scanner, entry):
        return {'action_type': cls.Stop}

    @transition(froms=['scan_product'], to='stop', sequence=1)
    def transition_stop(self, session, scanner, entry):
        return entry == self.stop_code

    @transition(froms=['scan_product', 'scan_another_product'], to='scan_location', sequence=2)
    def transition_product_ok(self, session, scanner, entry):
        query = session.query([self.ProductLocation])
        query = query.filter(self.ProductLocation.job == scanner.job)
        query = query.filter(self.ProductLocation.product == entry)
        query = query.filter(self.ProductLocation.quantity_count < self.ProductLocation.quantity)
        return query.count() > 0

    @transition(froms=['scan_product', 'scan_another_product'], to='scan_another_product', sequence=3)
    def transition_product_ko(self, session, scanner, entry):
        return True

    @transition(froms=['scan_location'], to='scan_product', sequence=1)
    def transition_location_ok(self, session, scanner, entry):
        return scanner.properties['location'] == entry

    @transition(froms=['scan_location'], to='scan_location', sequence=2)
    def transition_product_ko(self, session, scanner, entry):
        return True

Jean-Sébastien Suzanne