🔍 Code Extractor

A class that generates formatted print messages for different types of laboratory objects (Parblock, Organ, Study, Slide, Reagent) by querying a Neo4j graph database and building a pipe-delimited message string.

Purpose

This class is responsible for creating print job messages for a label printing system in a laboratory information management context. It queries a Neo4j graph database to retrieve information about various laboratory objects and their relationships, then constructs a formatted message string containing relevant metadata. The class supports different object types (Parblock, Organ, Study, Slide, Reagent) and dynamically selects the appropriate printing method based on the object type. The generated message is intended to be sent to a print server for label generation.

Source Code

class print_client():    
    
    def __init__(self, graph, server, queue, labeltype, language, obj_uid, **kwargs):
        self._MAP = {
            "Parblock":self.print_block,
            "Organ":self.print_organ,
            "Study":self.print_study,
            "Slide":self.print_slide,
        }
        self.graph = graph
        self.message = ""
        
        self.add_server_information(server, queue, labeltype, language)
        self.add_dpmm(kwargs.get('dpmm', '8'))
        node = self.get_node(obj_uid)
        obj_type=list(node._labels)[0]
        self.add_objtype(obj_type)
        
        print_function = self._MAP.get(obj_type, self.print_other)
        print_function(node, obj_uid)
        
        
    def print_block(self, node, obj_uid):
        self.add_name(node)
        self.add_customer(obj_uid)
        self.add_organ_from_block(obj_uid)
        self.add_externalID(node)
        
    def print_organ(self, node, obj_uid):
        self.add_name(node)
        self.add_ID(obj_uid)
        self.add_UID(obj_uid)
        self.add_externalID(node)

    def print_study(self, node, obj_uid):
        self.add_name(node)
        self.add_customer(obj_uid)
        self.add_UID(obj_uid)
        self.add_externalID(node)
        
    def print_slide(self, node, obj_uid):
        self.add_customer(obj_uid)
        self.add_name(node)
        self.add_staining(node)
        self.add_externalID(node)
        self.add_externalID_study(obj_uid)
        
    def print_reagent(self, node, obj_uid, **kwargs):
        self.add_name(node)
        self.add_UID(obj_uid)
        self.add_batch(node)
        self.add_manufacturer(obj_uid)
        self.add_solution_or_mix(node)
        self.add_good_until_date(node)
        self.add_expiration_date(node)
        
    def print_other(self, node, obj_uid):
        self.add_name(node)
        self.add_UID(obj_uid)
        
    def add_server_information(self, server, queue, labeltype, language):
        self.message += f"server:{server}|queue:{queue}|labeltype:{labeltype}|language:{language}"
        
    def get_node(self, obj_uid):
        return self.graph.run(f"MATCH (o) WHERE o.UID = '{obj_uid}' RETURN o").evaluate()
    
    def add_batch(self, node):
        self.message+=f"|batch:{node.get('batch')}"

    def add_manufacturer(self, obj_uid):
        manufacturer = self.graph.run(f"MATCH (m:Manufacturer)-->(r:Reagent {{UID:'{obj_uid}'}}) RETURN m.N LIMIT 1").evaluate()
        if manufacturer == 'Connected-Pathology':
            manufacturer = 'Mixed Product'
        self.message+=f"|manufacturer:{manufacturer}"
        
    def add_good_until_date(self, node):
        self.message+=f"|good_until_date:{node.get('good_until_date').to_native().strftime('%Y-%m-%d')}" if node.get('good_until_date') else ""
    
    def add_expiration_date(self, node):
        self.message+=f"|expiration:{node.get('expiration').to_native().strftime('%Y-%m-%d')}" if node.get('expiration') else ""
        
    def add_solution_or_mix(self, node):
        parents = self.graph.run(f"MATCH (p:Reagent)-[rel]->(:Reagent {{UID:'{node['UID']}'}}) RETURN COLLECT(rel)").evaluate()
        if len(parents) == 0:
            return
        self.message+=f"|mix:"
        for parent in parents:
            self.message+=f"={parent.start_node['N']}@{parent.relationship['parts']}"
    
    def add_dpmm(self, dpmm):
        self.message+= f"|dpmm:{dpmm}"
    
    def add_ID(self, obj_uid):
        animal_ID = self.graph.run(f"MATCH (e:ExpItem)-->(o:Organ {{UID:'{obj_uid}'}}) RETURN DISTINCT e.N").evaluate()
        self.message += f"|ID:{animal_ID}"
        
    def add_organ_from_block(self, obj_uid):
        organ = self.graph.run(f"MATCH (o:Organ)-->(p:Parblock {{UID:'{obj_uid}'}}) RETURN DISTINCT o.N LIMIT 1").evaluate()
        self.message += f"|organ:{organ}" ##unused
    
    def add_customer(self, obj_uid):
        customer = self.graph.run(f"MATCH (c:Customer)-[*..7]->(o) WHERE o.UID = '{obj_uid}' RETURN DISTINCT CASE WHEN c.shorthand IS NOT NULL AND NOT isEmpty(c.shorthand) THEN c.shorthand ELSE c.N END").evaluate()
        self.message += f"|customer:{customer}"
        
    def add_name(self, node):
        if node.get('Alias'):
            name = node['N'] + ' (' + node['Alias'] + ')'
        else:
            name = node['N']
        self.message += f"|name:{name}"
    
    def add_UID(self, obj_uid):
        self.message += f"|UID:{obj_uid}"
        
    def add_staining(self, node):
        self.message += f"|staining:{node['T']}"
        
    def add_externalID(self, node):
        if node.get('external_N'):
            self.message += f"|external_N:{node['external_N']}"
            return
        external_n = self.graph.run(f"MATCH (o)-->(p) WHERE p.UID = '{node['UID']}' RETURN DISTINCT o.external_N LIMIT 1").evaluate()
        if external_n:
            self.message += f"|external_N:{external_n}"
            
    def add_externalID_study(self, obj_uid):
        external_n = self.graph.run(f"MATCH (s:Study)-[*..7]->(o) WHERE o.UID = '{obj_uid}' RETURN DISTINCT s.external_N").evaluate()
        self.message += f"|study_external_N:{external_n}"
        
    def add_objtype(self, obj_type):
        self.message+= f"|obj_type:{obj_type}"
        
    def send_job(self):
        # client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # client.connect(('print_server', 58888))
        # client.send(self.message.encode())
        # client.close()
        print(self.message)

Parameters

Name	Type	Default	Kind
bases	-	-

Parameter Details

graph: A Neo4j graph database connection object (py2neo Graph instance) used to execute Cypher queries to retrieve node and relationship data

server: String identifier for the print server destination (e.g., hostname or IP address)

queue: String identifier for the print queue to use on the print server

labeltype: String specifying the type/format of label to print (e.g., label template name)

language: String specifying the language for the label content (e.g., 'en', 'de')

obj_uid: String containing the unique identifier (UID) of the object to print a label for

**kwargs: Optional keyword arguments, currently supports 'dpmm' (dots per millimeter) with default value '8' for print resolution

Return Value

The constructor returns an instance of print_client with a fully constructed message string stored in the 'message' attribute. The send_job() method returns None but prints the message to stdout (socket communication is commented out).

Class Interface

Methods

Attributes

Name	Type	Description	Scope
_MAP	dict	Dictionary mapping object type strings to their corresponding print methods	instance
graph	Graph	Neo4j graph database connection object used for executing Cypher queries	instance
message	str	Pipe-delimited string containing all formatted print information, built incrementally during initialization	instance

Dependencies

• socket
• asyncio
• py2neo

Required Imports

import socket
import asyncio

Usage Example

from py2neo import Graph

# Connect to Neo4j database
graph = Graph('bolt://localhost:7687', auth=('neo4j', 'password'))

# Create print client for a Parblock object
printer = print_client(
    graph=graph,
    server='print_server_hostname',
    queue='label_queue_1',
    labeltype='standard_label',
    language='en',
    obj_uid='BLOCK-12345',
    dpmm='12'
)

# Send the print job (currently prints to stdout)
printer.send_job()

# Access the generated message
print(printer.message)

Best Practices

• Ensure the Neo4j graph database connection is active before instantiating the class
• Verify that the obj_uid exists in the database before creating a print_client instance to avoid None node errors
• The class automatically constructs the message during initialization, so instantiation triggers all database queries
• The message attribute is built incrementally and should not be modified directly after instantiation
• The send_job() method currently only prints to stdout; uncomment socket code for actual network printing
• Handle potential database query failures as the class does not include error handling for missing nodes or relationships
• The class assumes specific graph schema with particular node labels and properties; ensure database schema matches expectations
• Date formatting assumes node properties contain date objects with to_native() method (Neo4j temporal types)
• The _MAP dictionary determines which print method is called; unsupported object types fall back to print_other()

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class TSC_Label 65.5% similar

  A class for creating and printing TSCPL2 format labels for laboratory information system (LIS) objects, specifically designed for TSC label printers with support for DataMatrix barcodes and text formatting.

  From: /tf/active/vicechatdev/resources/printers.py

• function main_v117 55.0% similar

  Initializes a TCP messenger client and sends a job to a server queue for processing graph-related tasks with specified label types and language settings.

  From: /tf/active/vicechatdev/resources/printclient.py

• class Cassette_Printer 54.0% similar

  A class that generates and saves cassette label images for printing by writing cassette information to a shared folder monitored by a Windows virtual PC.

  From: /tf/active/vicechatdev/resources/printers.py

• class Study_overview 51.9% similar

  A class that generates comprehensive study overview reports by querying a Neo4j graph database and producing Excel files with ID mappings, audit logs, and Gantt chart visualizations of study progress.

  From: /tf/active/vicechatdev/resources/documents.py

• class NodeLabels_v1 51.1% similar

  A constants class that defines string labels for Neo4j graph database node types used in a controlled document management system (CDocs).

  From: /tf/active/vicechatdev/CDocs single class/db/schema_manager.py

class print_client():    
    
    def __init__(self, graph, server, queue, labeltype, language, obj_uid, **kwargs):
        self._MAP = {
            "Parblock":self.print_block,
            "Organ":self.print_organ,
            "Study":self.print_study,
            "Slide":self.print_slide,
        }
        self.graph = graph
        self.message = ""
        
        self.add_server_information(server, queue, labeltype, language)
        self.add_dpmm(kwargs.get('dpmm', '8'))
        node = self.get_node(obj_uid)
        obj_type=list(node._labels)[0]
        self.add_objtype(obj_type)
        
        print_function = self._MAP.get(obj_type, self.print_other)
        print_function(node, obj_uid)
        
        
    def print_block(self, node, obj_uid):
        self.add_name(node)
        self.add_customer(obj_uid)
        self.add_organ_from_block(obj_uid)
        self.add_externalID(node)
        
    def print_organ(self, node, obj_uid):
        self.add_name(node)
        self.add_ID(obj_uid)
        self.add_UID(obj_uid)
        self.add_externalID(node)

    def print_study(self, node, obj_uid):
        self.add_name(node)
        self.add_customer(obj_uid)
        self.add_UID(obj_uid)
        self.add_externalID(node)
        
    def print_slide(self, node, obj_uid):
        self.add_customer(obj_uid)
        self.add_name(node)
        self.add_staining(node)
        self.add_externalID(node)
        self.add_externalID_study(obj_uid)
        
    def print_reagent(self, node, obj_uid, **kwargs):
        self.add_name(node)
        self.add_UID(obj_uid)
        self.add_batch(node)
        self.add_manufacturer(obj_uid)
        self.add_solution_or_mix(node)
        self.add_good_until_date(node)
        self.add_expiration_date(node)
        
    def print_other(self, node, obj_uid):
        self.add_name(node)
        self.add_UID(obj_uid)
        
    def add_server_information(self, server, queue, labeltype, language):
        self.message += f"server:{server}|queue:{queue}|labeltype:{labeltype}|language:{language}"
        
    def get_node(self, obj_uid):
        return self.graph.run(f"MATCH (o) WHERE o.UID = '{obj_uid}' RETURN o").evaluate()
    
    def add_batch(self, node):
        self.message+=f"|batch:{node.get('batch')}"

    def add_manufacturer(self, obj_uid):
        manufacturer = self.graph.run(f"MATCH (m:Manufacturer)-->(r:Reagent {{UID:'{obj_uid}'}}) RETURN m.N LIMIT 1").evaluate()
        if manufacturer == 'Connected-Pathology':
            manufacturer = 'Mixed Product'
        self.message+=f"|manufacturer:{manufacturer}"
        
    def add_good_until_date(self, node):
        self.message+=f"|good_until_date:{node.get('good_until_date').to_native().strftime('%Y-%m-%d')}" if node.get('good_until_date') else ""
    
    def add_expiration_date(self, node):
        self.message+=f"|expiration:{node.get('expiration').to_native().strftime('%Y-%m-%d')}" if node.get('expiration') else ""
        
    def add_solution_or_mix(self, node):
        parents = self.graph.run(f"MATCH (p:Reagent)-[rel]->(:Reagent {{UID:'{node['UID']}'}}) RETURN COLLECT(rel)").evaluate()
        if len(parents) == 0:
            return
        self.message+=f"|mix:"
        for parent in parents:
            self.message+=f"={parent.start_node['N']}@{parent.relationship['parts']}"
    
    def add_dpmm(self, dpmm):
        self.message+= f"|dpmm:{dpmm}"
    
    def add_ID(self, obj_uid):
        animal_ID = self.graph.run(f"MATCH (e:ExpItem)-->(o:Organ {{UID:'{obj_uid}'}}) RETURN DISTINCT e.N").evaluate()
        self.message += f"|ID:{animal_ID}"
        
    def add_organ_from_block(self, obj_uid):
        organ = self.graph.run(f"MATCH (o:Organ)-->(p:Parblock {{UID:'{obj_uid}'}}) RETURN DISTINCT o.N LIMIT 1").evaluate()
        self.message += f"|organ:{organ}" ##unused
    
    def add_customer(self, obj_uid):
        customer = self.graph.run(f"MATCH (c:Customer)-[*..7]->(o) WHERE o.UID = '{obj_uid}' RETURN DISTINCT CASE WHEN c.shorthand IS NOT NULL AND NOT isEmpty(c.shorthand) THEN c.shorthand ELSE c.N END").evaluate()
        self.message += f"|customer:{customer}"
        
    def add_name(self, node):
        if node.get('Alias'):
            name = node['N'] + ' (' + node['Alias'] + ')'
        else:
            name = node['N']
        self.message += f"|name:{name}"
    
    def add_UID(self, obj_uid):
        self.message += f"|UID:{obj_uid}"
        
    def add_staining(self, node):
        self.message += f"|staining:{node['T']}"
        
    def add_externalID(self, node):
        if node.get('external_N'):
            self.message += f"|external_N:{node['external_N']}"
            return
        external_n = self.graph.run(f"MATCH (o)-->(p) WHERE p.UID = '{node['UID']}' RETURN DISTINCT o.external_N LIMIT 1").evaluate()
        if external_n:
            self.message += f"|external_N:{external_n}"
            
    def add_externalID_study(self, obj_uid):
        external_n = self.graph.run(f"MATCH (s:Study)-[*..7]->(o) WHERE o.UID = '{obj_uid}' RETURN DISTINCT s.external_N").evaluate()
        self.message += f"|study_external_N:{external_n}"
        
    def add_objtype(self, obj_type):
        self.message+= f"|obj_type:{obj_type}"
        
    def send_job(self):
        # client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # client.connect(('print_server', 58888))
        # client.send(self.message.encode())
        # client.close()
        print(self.message)