path: root/src/test

#!/usr/bin/env python3
'''
pylibcoopgamma -- Python library for interfacing with cooperative gamma servers
Copyright (C) 2016  Mattias Andrée (maandree@kth.se)

This library is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this library.  If not, see <http://www.gnu.org/licenses/>.
'''

import os, sys, time, errno, select

os.chdir('/'.join(sys.argv[0].split('/')[:-1]))
sys.path.append('../bin')

import libcoopgamma
cg = libcoopgamma

if len(sys.argv) == 1:
    
    print('\033[1m%s:\033[m' % 'Methods')
    for m in cg.get_methods():
        print(m)
    print()
    
    print('\033[1m%s:\033[m' % 'Method')
    print(cg.get_method_and_site()[0])
    print()
    
    print('\033[1m%s:\033[m' % 'Site')
    print(cg.get_method_and_site()[1])
    print()
    
    print('\033[1m%s:\033[m' % 'PID file')
    print(cg.get_pid_file())
    print()
    
    print('\033[1m%s:\033[m' % 'Socket')
    print(cg.get_socket_file())
    print()
    
    g = cg.Context()
    g.connect()
    g.detach()
    gstr = repr(g)
    del g
    
    argv0 = './' + sys.argv[0].split('/')[-1]
    os.execl(argv0, argv0, gstr)
else:
    g = eval(sys.argv[1])
    g.attach()
    pass

print('\033[1m%s:\033[m' % 'CRTC:s')
crtcs = g.get_crtcs_sync()
for crtc in crtcs:
    print(crtc)
print()

info = g.get_gamma_info_sync(crtc)
print('\033[1m%s:\033[m' % 'CRTC info')
print('Cooperative:', 'yes' if info.cooperative else 'no')
if info.depth is not None:
    print('Depth:', cg.Depth.str(info.depth))
print('Supported:', cg.Support.str(info.supported))
if info.red_size is not None:
    print('Red stops:', info.red_size)
if info.green_size is not None:
    print('Green stops:', info.green_size)
if info.blue_size is not None:
    print('Blue stops:', info.blue_size)
print('Colourspace:', cg.Colourspace.str(info.colourspace))
if info.gamut is not None:
    print('Red point:', str(info.gamut.red))
    print('Green point:', str(info.gamut.green))
    print('Blue point:', str(info.gamut.blue))
    print('White point:', str(info.gamut.white))
print()

table = g.get_gamma_sync(cg.FilterQuery(crtc = crtc, coalesce = False))
print('\033[1m%s:\033[m' % 'Filter table')
print('Red stops:', table.red_size)
print('Green stops:', table.green_size)
print('Blue stops:', table.blue_size)
print('Depth:', cg.Depth.str(table.depth))
for i, fltr in enumerate(table.filters):
    print('Filter %i:' % i)
    print('  Priority:', fltr.priority)
    print('  Class:', fltr.fclass)
    print('  Ramps:')
    rr, gr, br = fltr.ramps.red, fltr.ramps.green, fltr.ramps.blue
    n = max(len(rr), len(gr), len(br))
    fmt = '    \033[31m%s \033[32m%s \033[34m%s\033[m'
    rr = [str(rr[i]) if i < len(rr) else '' for i in range(n)]
    gr = [str(gr[i]) if i < len(gr) else '' for i in range(n)]
    br = [str(br[i]) if i < len(br) else '' for i in range(n)]
    for y in zip(rr, gr, br):
        print(fmt % y)
print()
table_nc = table

table = g.get_gamma_sync(cg.FilterQuery(crtc = crtc, coalesce = True))
print('\033[1m%s:\033[m' % 'Filter table')
print('Red stops:', table.red_size)
print('Green stops:', table.green_size)
print('Blue stops:', table.blue_size)
print('Depth:', cg.Depth.str(table.depth))
for fltr in table.filters:
    print('Ramps:')
    rr, gr, br = fltr.ramps.red, fltr.ramps.green, fltr.ramps.blue
    n = max(len(rr), len(gr), len(br))
    fmt = '  \033[31m%s \033[32m%s \033[34m%s\033[m'
    rr = [str(rr[i]) if i < len(rr) else '' for i in range(n)]
    gr = [str(gr[i]) if i < len(gr) else '' for i in range(n)]
    br = [str(br[i]) if i < len(br) else '' for i in range(n)]
    for y in zip(rr, gr, br):
        print(fmt % y)
print()
table_c = table

fltr = cg.Filter(0, crtc, 'pylibcoopgamma::test::test', cg.Lifespan.UNTIL_DEATH, table.depth,
                 cg.Ramps(table.red_size, table.green_size, table.blue_size))

if table.depth < 0:
    Y = lambda x : x
else:
    m = 2 ** table.depth - 1
    Y = lambda x : int(x * m)

redzero = fltr.ramps.red
greenzero = fltr.ramps.green
redid = [Y(x / (table.red_size - 1)) for x in range(table.red_size)]
greenid = [Y(x / (table.green_size - 1)) for x in range(table.green_size)]
blueid = [Y(x / (table.blue_size - 1)) for x in range(table.blue_size)]

fltr.ramps.red = redid
g.set_gamma_sync(fltr)
time.sleep(0.5)
fltr.ramps.red = redzero

fltr.ramps.green = greenid
g.set_gamma_sync(fltr)
time.sleep(0.5)
fltr.ramps.green = greenzero

fltr.ramps.blue = blueid
g.set_gamma_sync(fltr)
time.sleep(0.5)

g.set_nonbreaking(False)

asynch = cg.AsyncContext()

def flush(e):
    if e.errno in (errno.EINTR, errno.EWOULDBLOCK, errno.EAGAIN):
        while True:
            try:
                g.flush()
            except OSError as ex:
                if ex.errno in (errno.EINTR, errno.EWOULDBLOCK, errno.EAGAIN):
                    continue
                else:
                    raise ex
            break
    else:
        raise e

try:
    g.get_crtcs_send(asynch)
except OSError as e:
    flush(e)
if g.synchronise([asynch]) != 0:
     sys.exit(1)
if g.get_crtcs_recv(asynch) != crtcs:
    sys.exit(2)

try:
    g.get_gamma_info_send(crtc, asynch)
except OSError as e:
    flush(e)
if g.synchronise([asynch]) != 0:
     sys.exit(3)
info2 = g.get_gamma_info_recv(asynch)
if info2.cooperative != info.cooperative:
    sys.exit(4)
if info2.depth != info.depth:
    sys.exit(4)
if info2.supported != info.supported:
    sys.exit(4)
if info2.red_size != info.red_size:
    sys.exit(4)
if info2.green_size != info.green_size:
    sys.exit(4)
if info2.blue_size != info.blue_size:
    sys.exit(4)
if info2.colourspace != info.colourspace:
    sys.exit(4)
if (info2.gamut is None) != (info.gamut is None):
    sys.exit(4)
if info.gamut is not None:
    if info2.gamut.red.x_raw != info.gamut.red.x_raw or info2.gamut.red.x != info.gamut.red.x:
        sys.exit(4)
    if info2.gamut.red.y_raw != info.gamut.red.y_raw or info2.gamut.red.y != info.gamut.red.y:
        sys.exit(4)
    if info2.gamut.green.x_raw != info.gamut.green.x_raw or info2.gamut.green.x != info.gamut.green.x:
        sys.exit(4)
    if info2.gamut.green.y_raw != info.gamut.green.y_raw or info2.gamut.green.y != info.gamut.green.y:
        sys.exit(4)
    if info2.gamut.blue.x_raw != info.gamut.blue.x_raw or info2.gamut.blue.x != info.gamut.blue.x:
        sys.exit(4)
    if info2.gamut.blue.y_raw != info.gamut.blue.y_raw or info2.gamut.blue.y != info.gamut.blue.y:
        sys.exit(4)
    if info2.gamut.white.x_raw != info.gamut.white.x_raw or info2.gamut.white.x != info.gamut.white.x:
        sys.exit(4)
    if info2.gamut.white.y_raw != info.gamut.white.y_raw or info2.gamut.white.y != info.gamut.white.y:
        sys.exit(4)

try:
    g.set_gamma_send(cg.Filter(crtc = crtc, fclass = 'pylibcoopgamma::test::test',
                               lifespan = cg.Lifespan.REMOVE), asynch)
except OSError as e:
    flush(e)
if g.synchronise([asynch]) != 0:
     sys.exit(1)
g.set_gamma_recv(asynch)

try:
    g.get_gamma_send(cg.FilterQuery(crtc = crtc, coalesce = False), asynch)
except OSError as e:
    flush(e)
if g.synchronise([asynch]) != 0:
     sys.exit(5)
table = g.get_gamma_recv(asynch)
if table.red_size != table_nc.red_size:
    sys.exit(6)
if table.green_size != table_nc.green_size:
    sys.exit(6)
if table.blue_size != table_nc.blue_size:
    sys.exit(6)
if table.depth != table_nc.depth:
    sys.exit(6)
if len(table.filters) != len(table_nc.filters):
    sys.exit(6)
for i, (fltr1, fltr2) in enumerate(zip(table.filters, table_nc.filters)):
    if fltr1.priority != fltr2.priority:
        sys.exit(6)
    if fltr1.fclass != fltr2.fclass:
        sys.exit(6)
    if fltr1.ramps.red != fltr2.ramps.red:
        sys.exit(6)
    if fltr1.ramps.green != fltr2.ramps.green:
        sys.exit(6)
    if fltr1.ramps.blue != fltr2.ramps.blue:
        sys.exit(6)

try:
    g.get_gamma_send(cg.FilterQuery(crtc = crtc, coalesce = True), asynch)
except OSError as e:
    flush(e)
if g.synchronise([asynch]) != 0:
     sys.exit(7)
table = g.get_gamma_recv(asynch)
if table.red_size != table_c.red_size:
    sys.exit(8)
if table.green_size != table_c.green_size:
    sys.exit(8)
if table.blue_size != table_c.blue_size:
    sys.exit(8)
if table.depth != table_c.depth:
    sys.exit(8)
if len(table.filters) != len(table_c.filters):
    sys.exit(8)
for i, (fltr1, fltr2) in enumerate(zip(table.filters, table_c.filters)):
    if fltr1.ramps.red != fltr2.ramps.red:
        sys.exit(8)
    if fltr1.ramps.green != fltr2.ramps.green:
        sys.exit(8)
    if fltr1.ramps.blue != fltr2.ramps.blue:
        sys.exit(8)


fltr = cg.Filter(0, crtc, 'pylibcoopgamma::test::test', cg.Lifespan.UNTIL_DEATH, table.depth,
                 cg.Ramps(table.red_size, table.green_size, table.blue_size))

fltr.ramps.red = list(reversed(redid))
fltr.ramps.green = list(reversed(greenid))
fltr.ramps.blue = list(reversed(blueid))
try:
    g.set_gamma_send(fltr, asynch)
except OSError as e:
    flush(e)
if g.synchronise([asynch]) != 0:
    sys.exit(9)
g.skip_message()
time.sleep(0.5)

fltr.ramps.red = redid
fltr.ramps.green = greenid
fltr.ramps.blue = blueid
g.set_gamma_sync(fltr)

g.set_nonbreaking(True)

async1 = asynch
async2 = cg.AsyncContext()
async3 = cg.AsyncContext()
time.sleep(0.5)

fltr.ramps.red = list(reversed(redid))
fltr.fclass = 'pylibcoopgamma::test::red'
try:
    g.set_gamma_send(fltr, async1)
    need_flush = True
except OSError as e:
    if e.errno not in (errno.EINTR, errno.EWOULDBLOCK, errno.EAGAIN):
        sys.exit(10)
    need_flush = False
fltr.ramps.red = redid

fltr.ramps.green = list(reversed(greenid))
fltr.fclass = 'pylibcoopgamma::test::green'
try:
    g.set_gamma_send(fltr, async2)
    need_flush = True
except OSError as e:
    if e.errno not in (errno.EINTR, errno.EWOULDBLOCK, errno.EAGAIN):
        sys.exit(10)
    need_flush = False
fltr.ramps.green = greenid

fltr.ramps.blue = list(reversed(blueid))
fltr.fclass = 'pylibcoopgamma::test::blue'
try:
    g.set_gamma_send(fltr, async3)
    need_flush = True
except OSError as e:
    if e.errno not in (errno.EINTR, errno.EWOULDBLOCK, errno.EAGAIN):
        sys.exit(10)
    need_flush = False
fltr.ramps.blue = blueid

while need_flush:
    try:
        g.flush()
        break
    except OSError as e:
        if e.errno in (errno.EINTR, errno.EWOULDBLOCK, errno.EAGAIN):
            continue
        else:
            sys.exit(11)

poll = select.poll()
poll.register(g.fd, select.POLLIN | select.POLLPRI)

unsynced = [True] * 3
while any(unsynced):
    poll.poll()
    n = g.synchronise([async1, async2, async3])
    if n < 0 or n >= 3 or not unsynced[n]:
        sys.exit(12)
    unsynced[n] = False
    g.set_gamma_recv([async1, async2, async3][n])

time.sleep(0.5)
del g