ActivePOS events received by the script are represented by objects that have the following fields:
Table 11.
| Field | Value |
|---|---|
| article | Article number |
| associated_channel | Video channel associated with POS terminal |
| barcode | Item barcode |
| cashier_name | Name of the cashier |
| discount | Total discount on check |
| discount_card | Discount card number |
| flags | Event flags |
| location | Location |
| op_id | Internal receipt number (may contain the actual receipt number) |
| pos_terminal | POS GUID |
| pos_terminal_name | POS name |
| position | Item in the receipt |
| price | Item price |
| price_per_unit | Price per unit |
| quantity | Item quantity |
| text | Text (item name, bonus card number, message, etc.) |
| ts_in_receipt | Time of the event generation on the terminal |
| ts_received | Time of event receipt by the server |
| type | Event type |
| weight | Item weight |
The activate_on_pos_events() function is used to get ActivePOS events
import time
def f(ev):
message("Unique event number: %s" % ev.op_id)
message("Event type: %s" % ev.type)
message("Terminal ID: %s" % ev.pos_terminal)
message("Terminal name: %s" % ev.pos_terminal_name)
message("Associated video channel: %s" % ev.associated_channel)
message("Flags: %s" % ev.flags)
message("Position number: %s" % ev.position)
message("Text: %s" % ev.text)
message("Price per unit: %0.2f" % (ev.price/100.0))
message("Weight: %0.3f" % (ev.weight/1000.0))
message("Quantity: %s" % ev.quantity)
message("Article: %s" % ev.article)
message("Barcode: %s" % ev.barcode)
message("Location: %s" % ev.location)
message("Time of arrival on server: %s" %
time.strftime("%H:%M:%S %d.%m.%Y",
time.localtime(ev.ts_received/1000000)))
message("Time indicated on receipt: %s" %
time.strftime("%H:%M:%S %d.%m.%Y",
time.localtime(ev.ts_in_receipt/1000000)))
activate_on_pos_events(f) Tip
The price is given in whole numbers in pennies, and the weight is given in grams. The reception time of the message may differ from the time recorded on the point-of-sale terminal. To find the required moment in the video archive, use the reception time of the event.
To search for events from POS terminals in the database, use the search_pos_events() function. The function accepts 4 parameters:
- callable object, which will be called for each event found;
- the lower limit of the time interval in which the events will be searched for;
- the upper limit of the time interval in which the events will be searched for;
- filter.
As a filter, you need to transmit a dictionary, in which you can use the following keys:
Table 12.
| Key | Value | Default value |
|---|---|---|
| cashiers | Names of cashiers in whose receipts you need to find the events | [ ] |
| events | Types of events to find | [ ] |
| receipt_number | Numbers of receipts in the receipts of which you need to find the events | [ ] |
| terminals | Terminals, in the receipts from which you want to find the events | [ ] |
| text | Text containing in events (event_text, event_article, event_barcode) that you need to find | [ ] |
import time
from pos_utils import *
filter = {
"receipt_number": "1234567",
"text": [StringFilter("Milk", SearchFlags.STARTS_WITH)],
"cashiers": ("Ivanov I",)
}
search_pos_events(lambda event: alert(event.text), time.time() - 24 * 60 * 60, time.time(), filter) This example will find all events (maximum: 500) in the checks with the number "1234567" created by the cashier "Ivanov I", in the event_text, event_article or event_barcode fields containing the text beginning with the string "Milk" and received in the last 24 hours. The text of each event will be displayed in a separate pop-up window.
A script can find suspicious situations. You can use the pos_fraud() function to attract the operator's attention and record an alarm event on a receipt. You can create a filter to search and highlight based on the presence of such event in a receipt.
import time
def f(ev):
if time.localtime().tm_hour < 23: return
if ev.type!="POS_POSITION_ADD": return
u = ev.text.decode("utf-8").upper().encode("utf-8")
for w in ["BEER", "WINE", "VODKA", "COGNAC"]:
if u.find(w) != -1:
pos_fraud(ev, "Alcohol after 11pm")
return
activate_on_pos_events(f) The upper() function converts the string to upper case (all capital letters). For this conversion to work, the string must be in Unicode (utf-8).
Incidents generated by ActivePOS detectors and received by the script are represented by objects with the following fields:
Table 13.
| Field | Value |
|---|---|
| associated_channel | Video channel associated with POS terminal |
| cashier_name | Name of the cashier |
| comment | Commentary on incident |
| custom_columns | Additional information |
| detector | GUID of the detector that created the incident |
| expert | Name of the operator who processed the incident |
| expert_estimate | Damage assessment |
| id | Unique incident ID |
| operator | Name of the operator who created the incident |
| related_ts | Times of the events that triggered the detector |
| review_duration | Duration of incident processing |
| review_ts | End time of incident processing |
| server | Server GUID |
| status | Incident status (0 - new, 1 - confirmed, 2 - rejected) |
| terminal_guid | POS GUID |
| terminal_name | POS name |
| ts_created | Incident creation time |
| type_description | Incident description |
| type_id | GUID of the incident type |
| type_name | Incident name |
The activate_on_pos_incidents() function is used to get messages about new incidents. The function accepts two parameters:
- Detector GUID;
- callable object, which will be called for each received ActivePos incident created by the detector with the given GUID.
Tip
If you send an empty string as GUID, the callable object will be called for any new incident.
activate_on_pos_incidents("", lambda incident: alert(incident.type_name)) The search_pos_incidents() function can be used to search for incidents in the database. The function accepts 4 parameters:
- callable object, which will be called for each ActivePos incident found;
- the lower limit of the time interval in which the incidents will be searched for;
- the upper limit of the time interval in which the incidents will be searched for;
- filter.
As a filter, you need to transmit a dictionary, in which you can use the following keys:
Table 14.
| Key | Value | Default value |
|---|---|---|
| cashiers | The cashiers in whose receipts you need to find incidents | [ ] |
| detectors_names | The names of the detectors that generated the incidents or the types of incidents to find | [ ] |
| experts | The operators who confirmed or rejected the incidents to be found | [ ] |
| flags | The bitmask of incident search flags:
|
0 |
| incidents_ids | Types of incidents to find | [ ] |
| limit | Maximum number of requested incidents | -1 |
| terminals | The terminals, in the receipts from which you need to find the incidents | [ ] |
import time
from pos_utils import IncidentFlag
filter = {
"flags": IncidentFlag.FILTER_CONFIRMED | IncidentFlag.FILTER_NONEMPTY_ESTIMATE,
"limit": 20,
}
search_pos_incidents(lambda incident: alert(incident.type_name), time.time() - 24 * 60 * 60 * 2, time.time(), filter) This example will analyze all the receipts stored in the database for the last 2 days, then find the last 20 confirmed incidents with a non-zero damage estimate among them, and display the name of each type of incident in a pop-up window.
The pos_process_archive() function can be used to start the processing of the event archive by a particular detector. The function has 3 arguments:
- Detector GUID;
- the lower limit of the time interval of the event archive;
- the upper limit of the time interval of the event archive.
import time
pos_process_archive("T1RuoVF7", time.time() - 24 * 60 * 60 * 2, time.time()) This example will start processing checks for the last 48 hours by a detector with a GUID equal to "T1RuoVF7".
The pos_incident_create() function can be used to create a new incident from a script. The function uses 4 mandatory arguments:
- The GUID of the terminal that will be associated with the created incident;
- name of the created incident type;
- name of the cashier;
- the function to which the result of the incident creation operation will be sent in case of an error.
pos_incident_create("ZyPx6vF0", "Cancel position without the administrator", "Ivanov I", lambda err: alert(err.msg)) This example will create an incident with the "Cancel position without the administrator" type, associated with the terminal, the GUID of which is "ZyPx6vF0", and the cashier "Ivanov I".
In order to transfer or restore the detector configurations, you can use the pos_import_detector() function. This function accepts data in xml format containing the detector configuration.
with open("pos_detectors.xml") as detectors_config:
pos_import_detector(detectors_config.read()) This example will delete the current detector configuration and restore it from the pos_detectors.xml file.
The generate_pos_report() function can be used to generate ActivePOS reports. The function has 4 arguments:
- report type;
- the lower limit of the time interval in which the data for the report will be searched;
- the upper limit of the time interval in which the data for the report will be searched;
- the function to which the object containing the report will be sent.
import time
from pos_utils import ReportType
def report_ready(res):
with open("pos_report.ods", "w") as report:
report.write(res.zipped_report)
generate_pos_report(ReportType.VIOLATIONS_REPORT, str(int((time.time() - 24 * 60 * 60 * 10) * 1e6)), str(int(time.time() * 1e6)), report_ready) This example generates a "Violation Report" for the last 10 days and saves it to the file pos_report.ods.
Tip
The module pos_utils.py contains a description of the StringFilter, SearchFlags, ReportType and IncidentFlag classes, and is located in the pyredist folder.
Tip
The time is in microseconds in UNIX time format based on the timezone configured on server.

