pyvlog is a Python package for working with traffic device data from smart intersections that adhere to the V-Log messaging protocol. It can be used to convert sequences of device messages into time-stamped statuses for any combination of traffic devices at a smart intersection.
It converts individual device messages like this:
012018091115000000
05000003000
0D00000200
090000020070A0
0E00310102
0600610201
into intersection statuses like this:
{'timestamp': 1536670800.6,
'tijdReferentie': 1536670800.0,
'detectie': {0: {'OG-BG-FL': 0, 'storing': 0, 'bezet': 0},
1: {'OG-BG-FL': 0, 'storing': 0, 'bezet': 0},
2: {'OG-BG-FL': 0, 'storing': 0, 'bezet': 1}},
'externeSignaalgroep': {0: 0, 1: 2},
'deltaTijd': 0.6}
Naming and data conventions for the various traffic devices follow the specification provided by Vialis.
pyvlog can be installed via pip:
pip install pyvlog
Documentation is available at readthedocs.
pyvlog is designed for converting sequences of v-log messages - streamed from smart intersection traffic devices - into time-stamped statuses of all traffic devices at an intersection. These statuses can be queried by both time and device, aiding the development of functionality built around v-log data.
pyvlog is able to perform this conversion in two ways.
parsers are objects which receive messages individually and use each message to update the stored intersection status. Additionally they can identify when a status is "complete" (all values are known for a specific timestamp) and can log these completed statuses. They are suited to converting realtime streams of v-log messages.
converters are functions which take a sequence of v-log messages and return a sequence of statuses. They are suited to (batch) converting data dumps of historic v-log messages.
The parsing of individual v-log messages is always done by the VLogParser class. At its simplest, you can create a VLogParser() object and call its .parse_message() method one message at a time on a sequence of messages. The object stores the intersection status as a dictionary in the attribute .status. This is the intersection status after the last message was received.
from pyvlog.parsers import VLogParser
vlogger = VLogParser()
messages = ['012018091115000000', '05000003000', '0D00000200', '090000020070A0', '0E00310102', '0600610201']
for m in messages:
vlogger.parse_message(m)
print(vlogger.status)When converting a sequence of v-log messages you may well want to know the status of the intersection for all timestamps spanned by the sequence of messages, not just the final status. The VLogParser class identifies when a status is complete (there is not a one-to-one correspondence between messages and statuses) and logs this status by calling the method .log_status().
The base VLogParser class does not store these logged statuses anywhere, however two child classes are provided, VLogParserToList and VLogParserToJson, which log the statuses to a list and a JSON file respectively.
from pyvlog.parsers import VLogParserToList
status_list = []
vlogger = VLogParserToList(status_list)
messages = ['012018091115000000', '05000003000', '0D00000200', '090000020070A0', '0E00310102', '0600610201']
for m in messages:
vlogger.parse_message(m)
print(status_list)Instructions on writing your own parser classes are provided below.
A number of converter functions are provided for converting complete sets of v-log messages into their corresponding time-stamped statuses. These convert from a file or list of v-log messages to a JSON file, list or pandas dataframe of historic statuses (e.g. file_to_file or list_to_dataframe). Status data is written in JSON format, using UltraJSON.
from pyvlog.converters import file_to_json
vlog_file = "test.vlg"
out_file = "test.json"
file_to_json(vlog_file, out_file)For conversion to pandas dataframes the status dictionary is flattened (with "_" joining the keys) and timing fields are converted to datetime / timedelta.
from pyvlog.converters import list_to_dataframe
messages = ['012018091115000000', '05000003000', '0D00000200', '090000020070A0', '0E00310102', '0600610201']
df = list_to_dataframe(messages)
print(df.head())Custom parser classes can be created for any number of different logging routines, simply by inheriting the base VLogParser class and defining a new .log_status() method, plus any additional arguments. The two additional classes defined in the parsers module, VLogParserToList and VLogParserToJson, illustrate how such a custom parsing class may be created.
This package is developed for the processing of realtime v-log messages from a small number of smart intersections. As such not all types of v-log messages were available during its development. The message types currently parsed are given by the keys of messagetypes.MESSAGE_TYPE_DICT and are repeated below (with the v-log message prefix given in brackets).
- tijdReferentie (1)
- vlogInformatie (4)
- detectie (5,6)
- overigeIngangen (7,8)
- interneFaseCyclus (9,10)
- overigeUitgangenGUS (11,12)
- externeSignaalgroep (13,14)
- overigeUitgangenWUS (15,16)
- gewensteProgrammaStatus (17,18)
- werkelijkeProgrammaStatus (19,20)
- thermometer (23,24)
- instructieVariabelen (32)
- OVHulpdienstInformatie (34)
By default only detectie and externeSignaalgroep are parsed (plus tijdReferentie, which is always parsed in order to calculate the timestamp). Both the parser classes and converter functions take an argument logged_types, which specifies which message types to log.
from pyvlog.parsers import VLogParser
vlogger = VLogParser(logged_types=["detectie", "externeSignaalgroep", "interneFaseCyclus"])
messages = ['012018091115000000', '05000003000', '0D00000200', '090000020070A0', '0E00310102', '0600610201']
for m in messages:
vlogger.parse_message(m)
print(vlogger.status)