|
import omero |
|
from omero.gateway import BlitzGateway |
|
from omero.sys import ParametersI |
|
import sys |
|
|
|
USERNAME = '' |
|
PASSWORD = '' |
|
HOST = 'localhost' |
|
PORT = 4064 |
|
|
|
|
|
def connect(): |
|
""" Create an OMERO Connection """ |
|
|
|
# Initialize the connection |
|
conn = BlitzGateway(USERNAME, |
|
PASSWORD, |
|
host=HOST, |
|
port=PORT) |
|
|
|
# Connect |
|
connected = conn.connect() |
|
|
|
# Check that the connection was established |
|
if not connected: |
|
sys.stderr.write("Error: Connection not available, " |
|
"please check your user name and password.\n") |
|
sys.exit(1) |
|
return conn |
|
|
|
|
|
def queryImageTags(image_id): |
|
""" Query for tag annotations attached to an image. This will also |
|
return details (owner, group, etc) about the annotation. |
|
""" |
|
conn = connect() |
|
|
|
qs = conn.getQueryService() |
|
|
|
conn.SERVICE_OPTS.setOmeroGroup(-1) |
|
|
|
params = ParametersI() |
|
params.add('iid', omero.rtypes.wrap(image_id)) |
|
|
|
q = """ |
|
SELECT anno |
|
FROM Image image |
|
JOIN image.annotationLinks links |
|
JOIN links.child anno |
|
WHERE image.id = :iid |
|
AND anno.class = TagAnnotation |
|
""" |
|
|
|
for row in qs.projection(q, params, conn.SERVICE_OPTS): |
|
print 'Tag: %s (%s)' % (row[0].val.getTextValue().val, |
|
row[0].val.getDescription().val) |
|
|
|
|
|
def queryImagesTags(): |
|
""" Query for all tag annotations attached to images. This will also |
|
return details (owner, group, etc) about the annotation. |
|
""" |
|
conn = connect() |
|
|
|
qs = conn.getQueryService() |
|
|
|
conn.SERVICE_OPTS.setOmeroGroup(-1) |
|
|
|
params = ParametersI() |
|
|
|
q = """ |
|
SELECT image, anno |
|
FROM Image image |
|
JOIN image.annotationLinks links |
|
JOIN links.child anno |
|
WHERE anno.class = TagAnnotation |
|
""" |
|
|
|
for row in qs.projection(q, params, conn.SERVICE_OPTS): |
|
image = row[0].val |
|
anno = row[1].val |
|
print 'Image: %s Tag: %s (%s)' % (image.getName().val, |
|
anno.getTextValue().val, |
|
anno.getDescription().val) |
|
|
|
|
|
def queryImageTagsTerse(image_id): |
|
""" Query for tag annotation details only, attached to an image. |
|
""" |
|
conn = connect() |
|
|
|
qs = conn.getQueryService() |
|
|
|
conn.SERVICE_OPTS.setOmeroGroup(-1) |
|
|
|
params = ParametersI() |
|
params.add('iid', omero.rtypes.wrap(image_id)) |
|
|
|
q = """ |
|
SELECT anno.textValue, anno.description |
|
FROM Image image |
|
JOIN image.annotationLinks links |
|
JOIN links.child anno |
|
WHERE image.id = :iid |
|
AND anno.class = TagAnnotation |
|
""" |
|
|
|
for row in qs.projection(q, params, conn.SERVICE_OPTS): |
|
print 'Tag: %s (%s)' % (row[0].val, |
|
row[1].val) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
|
# Must be a long |
|
queryImageTags(103L) |
|
|
|
queryImagesTags() |
|
|
|
# Must be a long |
|
queryImageTagsTerse(103L) |
