lire 2:2.1.1-2.1 / usr / share / perl5 / Lire / Group.pm

package Lire::Group;

use strict;

use base qw/ Lire::Aggregator /;

use Carp;

use Lire::DlfQuery;
use Lire::ReportOperator qw/ group_data_value /;
use Lire::DataTypes qw/ check_int is_numeric_type /;

=pod

=head1 NAME

Lire::Group - Base class for implementation of the group aggregator

=head1 SYNOPSIS

    use Lire::Group;

=head1 DESCRIPTION

This module is the base class for implementation of the group
aggregator. This aggregator will split the DLF records in groups having
the same key. The key is made up of one or more several non-numerical
fields. The number of keys to keep in the report can be configured
through the limit attribute. The keys can also be sorted using the
sort attribute.

=head1 CONSTRUCTOR

=head2 new( %params )

Creates a new instance of a group aggregator. In addition to the
normal report operator parameters, the limit attribute can be
initialized at creation time.

=cut

sub new {
    my $proto = shift;
    my $class = ref $proto || $proto;

    my $self = bless { 'fields' => [] }, $class;

    my %args = @_;
    $self->SUPER::init( @_, 'op' => 'group' );

    $self->limit( $args{'limit'} )
      if exists $args{'limit'};

    return $self;
}

=pod

=head1 METHODS

=head2 limit( [$new_limit] )

Returns the number of keys to display in the report. This can be an
integer or the name of one of the report specification parameter (when
it starts by $).

The limit attribute can be changed by passing a $new_limit parameter.

=cut

sub limit {
    my ( $self, $limit ) = @_;

    if ( @_ == 2 ) {
	if ( defined $limit && $limit =~ /^\$/ ) {
	    my $name = substr $limit, 1;
	    croak "$limit isn't a defined parameter"
	      unless $self->report_spec->has_param( $name );
	    my $type = $self->report_spec->param( $name )->type();
	    croak "$limit parameter isn't of type int"
	      unless $type eq "int";
	    $self->{'limit'} = $limit;
	} elsif ( defined $limit ) {
	    croak "limit must be a positive integer"
	      unless check_int( $limit );
	    $self->{'limit'} = $limit;
	} else {
	    delete $self->{'limit'};
	}
    }
    return $self->{'limit'};
}

sub _max_entries {
    my $self = $_[0];

    my $ratio;

    my $limit = $self->report_spec->resolve_param_ref( $self->{'limit'} );
    return 0 unless ( $limit );

    if ( $limit < 10 ) {
	$ratio = 2;
    } elsif ( $limit < 25 ) {
	$ratio = 1.5
    } elsif ( $limit < 50 ) {
	$ratio = 1.25
    } elsif ( $limit < 1000 ) {
	$ratio = 1.1
    } else {
	$ratio = 1.05;
    }

    return int( $limit * $ratio );
}

=pod

=head2 sort_fields( [$new_sort_fields] )

Returns fields and/or operator values which are going to be used to
sort the entries generated by this aggregator. This is an array
reference containing group field names and/or operator's name.

The sorting fields can be changed by passing a new array reference.

=cut

sub sort_fields {
    my ( $self, $fields ) = @_;

    if (@_ == 2 && defined $fields ) {
	croak "$fields isn't an array reference"
	  unless UNIVERSAL::isa( $fields, "ARRAY" );

	foreach my $f ( @$fields ) {
	    croak "$f isn't a defined sort field name"
	      unless $self->is_valid_sort_field( $f );
	}
	if ( @$fields ) {
	    $self->{'sort_fields'} = $fields;
	} else {
	    delete $self->{'sort_fields'};
	}
    } elsif ( @_ == 2 ) {
	delete $self->{'sort_fields'};
    }

    return $self->{'sort_fields'};
}

=pod

=head2 group_fields( [$new_fields] )

Returns as a reference to an array of Lire::GroupField objects the DLF
fields that are used to group the records.

If the $new_fields is set, the DLF fields used to group the records
will be changed. This parameter should be a reference to an array of
Lire::GroupField objects.

=cut

sub group_fields {
    my ( $self, $fields ) = @_;

    if ( @_ == 2 ) {
	croak "$fields isn't an array reference"
	  unless UNIVERSAL::isa( $fields, "ARRAY" );

	croak "group fields array is empty"
	  unless @$fields;

	foreach my $f ( @$fields ) {
	    croak "$f isn't of type Lire::GroupField"
	      unless UNIVERSAL::isa( $f, "Lire::GroupField" );
	}
	$self->{'fields'} = $fields;
    }
    return $self->{'fields'};
}

# Overrides Lire::Aggregator::build_query
sub build_query {
    my ( $self, $query ) = @_;

    $self->SUPER::build_query( $query );

    foreach my $field ( @{$self->{'fields'}} ) {
        $query->add_group_field( $field->name );
    }

    $query->set_sort_spec( join( " ", @{$self->sort_fields()} ) )
      if $self->sort_fields();

    return;
}

# Overrides Lire::Aggregator::set_group_summary
sub set_group_summary {
    my ( $self, $group, $row ) = @_;

    $self->SUPER::set_group_summary( $group, $row );

    $group->show( $self->report_spec->resolve_param_ref( $self->{'limit'} ) )
      if defined $self->{'limit'};

    return;
}

# Implements Lire::Aggregator::create_entry
sub create_entry {
    my ( $self, $group, $row ) = @_;

    my $max_entries = $self->_max_entries();
    return undef if $max_entries && $group->entries() >= $max_entries;

    my @names = ();
    foreach my $field ( @{$self->{'fields'}} ) {
        my $value = $row->{ $field->name() };
        unless ( defined $value ) {
            $group->missing_cases( $group->missing_cases() +
                                   $row->{'_lr_nrecords'} );
            return undef;
        }
        push @names, $row->{ $field->name() };
    }
    if ( @names ) {
        my $entry = $group->create_entry();
        foreach my $n ( @names ) {
            $entry->add_name( $n );
        }
        return $entry;
    }
    # Not reached
}

# ------------------------------------------------------------------------
# Method is_name_defined()
#
# Overide Lire:::Aggregator one to also check among the group fields
sub is_name_defined {
    my ( $self, $name ) = @_;

    # Check in fields
    foreach my $f ( @{$self->{'fields'}} ) {
	return 1 if $f->name eq $name;
    }

    # Chain up
    $self->SUPER::is_name_defined( $name );
}

#------------------------------------------------------------------------
# Method is_valid_sort_field( $name )
#
# Method that checks if the name is valid as a sorting field. It is
# only if the name is the name of one of the operator in this group.
# It can optionnally be prefixed by - to specify descending sort
# order.
sub is_valid_sort_field {
    my ( $self, $name ) = @_;

    # Remove descending control char
    $name =~ s/^-//g;

    $self->is_name_defined( $name );
}

# ------------------------------------------------------------------------
# Method xml_attrs()
#
# Implementation required by Lire::Aggregator
sub xml_attrs {
    my ( $self ) = @_;

    my $attr = "";
    if ( defined $self->{'sort_fields'} &&
	 @{$self->{'sort_fields'}})
    {
	my $fields = join " ", @{$self->{'sort_fields'}};
	$attr .= qq{ sort="$fields"};
    }
    $attr .= qq{ limit="$self->{'limit'}"}
      if exists $self->{'limit'};

    $attr;
}

# ------------------------------------------------------------------------
# Method print_content( $fh, $pfx )
#
# Override to output the field elements.
sub print_content {
    my ( $self, $fh, $pfx ) = @_;

    foreach my $f ( @{$self->{'fields'}} ) {
	$f->print( $fh, $pfx + 1);
    }

    $self->SUPER::print_content( $fh, $pfx );
}

# Implements Lire::ReportOperator::name()
sub name {
    return 'group:' . join( ".", map { $_->name() } @{$_[0]->group_fields()} );
}

# Implements Lire::Aggregator::create_categorical_info
sub create_categorical_info {
    my ( $self, $info ) = @_;

    foreach my $field ( @{$self->group_fields} ) {
	$field->create_categorical_info( $info );
    }
}

# Implements Lire::ReportOperator::init_merge()
sub init_merge {
    my $self = $_[0];

    $self->SUPER::init_merge();

    # We can't use the array data structure for merging
    # because we aren't seeing the entries sorted.
    $self->{'use_array'} = 0;

    if ( defined $self->limit() ) {
	$self->{'limit_view'} =
          $self->report_spec()->resolve_param_ref( $self->limit() );

	# For better report merging, we add some entries.
	$self->{'limit_num'} = $self->_max_entries();
    }

    $self->{'sort_cmp'} = $self->make_sort_function();

    return;
}

#------------------------------------------------------------------------
# Method make_sort_function()
#
# Create a sort function which is used to sort the data according
# to the sort specification.
sub make_sort_function {
    my ( $self ) = @_;

    return unless $self->sort_fields;

    # Build sort function
    my ( $a_dflt, $b_dflt, $schwartzian );
    if (  $self->{'use_array'}) {
	$a_dflt = '$_[0]';
	$b_dflt = '$_[1]';
	$schwartzian = ''; # Not using schwarzian transform
    } else {
	$a_dflt = '$a';
	$b_dflt = '$b';
	$schwartzian = '[1]'; # Using schwartzian transform
    }

    my @sort_ops = ();
    foreach my $f ( @{$self->sort_fields} ) {
	my ($a, $b) = ($a_dflt, $b_dflt );
	if ( $f =~ /^-/ ) {
	    $f = substr $f, 1;
	    ($a,$b) = ($b,$a);
	}

	my $index;	# This will contains the index of the field in the array
	my $summary = ""; # This will be used to get at the summary data
	my $cmp = '<=>'; # Default to numeric comparison
	my $i = 0;
	foreach my $group_field ( @{$self->group_fields} ) {
	    if ( $group_field->name eq $f ) {
		$index = $i;
		if ( is_numeric_type( $group_field->field->type ) ) {
		    $cmp = "<=>";
		} else {
		    $cmp = "cmp";
		}
		last;
	    }
	    $i++;
	}

	# The sort field wasn't found in the group_fields,
	# look into the operators
	unless (defined $index) {
	    $i = @{$self->group_fields};
	    foreach my $op ( @{$self->ops} ) {
		# Skip aggregators
		next if $op->isa( 'Lire::Aggregator' );
		if ( $op->name eq $f ) {
		    $index = $i;
		    last;
		}
		$i++;
	    }
	}

	# Its wasn't found in one of our ReportOperator children, use
	# the summary value of one of our aggregator children
	unless (defined $index ) {
	    $i = @{$self->group_fields};
	    foreach my $op ( @{$self->ops} ) {
		# Only check aggregators
		next unless $op->isa( 'Lire::Aggregator' );
		if ( $op->is_name_defined($f) ) {
		    $index = $i;
		    $summary = $self->get_summary_value_string( $f );
		    last;
		}
		$i++;
	    }
	}
	croak "impossible to find $f value to sort on"
	  unless defined $index;
	if ( $summary ) {
	    push @sort_ops, 'group_data_value(' . $a ."->" . $schwartzian .
	      "[$index]$summary ) $cmp group_data_value( " . 
		$b ."->" . $schwartzian . "[$index]$summary)";
	} else {
	    push @sort_ops, $a ."->" . $schwartzian ."[$index] $cmp " .
	      $b ."->" . $schwartzian . "[$index]";
	}
    }
    my $sort_code = "sub { " . join( " || ", @sort_ops ) . " }";
    my $sort_func = eval $sort_code;
    croak "error compiling sort comparison ($sort_code): $@" if $@;

    $sort_func;
}

# Implements Lire::Aggregator::init_aggregator_data()
sub init_aggregator_data {
    my ( $self ) = @_;

    # The group datastructure used to hold the operations' data of the
    # group element is an array. It contains the fields value and the
    # op data: 
    # [ group field, group field, ..., op data, op data, op data ]

    if ( $self->{'use_array'}) {
	# To minimize memory, we operate on sorted input by keeping
	# the key in final sorted order and keeping only the minimum
	# needed
	#
	# Structure of the array: we keep in the first element the current 
	# group key, the second element is the current group element data.
	# After this we have the processed group elements data in sort order
	# and up the the limit attribute.
	# [ current_key, current_group_data, sorted_group_data, sorted_group_data, ... ]
	return [];
    } else {
	# For merging we don't see sorted input and keep related keys
	# using a hash.
	return {}
    }
}

sub item_data2sort_key {
   my $key = [];
   foreach my $f ( @{$_[0]} ) {
       push @$key, group_data_value( $f );
   }
   return $key;
}

# Implements Lire::Aggregator::merge_aggregator_data()
sub merge_aggregator_data {
    my ( $self, $group, $data ) = @_;

    foreach my $e ( $group->entries ) {
	my $key = join( ";", map { $_->{'value'} } $e->names );
	my $key_data = $data->{$key};
	unless ( $key_data ) {
	    $key_data = $data->{$key} = [];

	    my $i = 0;
	    my @names = $e->names;

	    croak "wrong number of names: expected ", 
	      scalar @{$self->group_fields}, " but found ", scalar @names
		if @names != @{$self->group_fields};

	    foreach my $f ( @{$self->group_fields} ) {
		$key_data->[$i] = $names[$i]{'value'};
		$i++;
	    }

	    foreach my $op ( @{$self->ops} ) {
		$key_data->[$i++] = $op->init_group_data();
	    }
	}

	my $i = @{$self->group_fields};
	foreach my $op ( @{$self->ops} ) {
	    my $value = $e->data_by_name( $op->name );
	    my $op_data = $key_data->[$i++];
	    $op->merge_group_data( $value, $op_data )
	      if ( $value );
	}
    }
}

=pod

=head2 binary_insert

Recursive function that uses a binary search to insert 
$item in $array using $cmp as comparison operator.
$first_idx and $max_idx specify the boundaries of the search.
Search ends when $first_idx == $max_idx or when $sort_key
sorts at or before $first_idx or at or after $max_idx

=cut

sub binary_insert {
    my ( $item, $sort_key, $array, $cmp, $first_idx, $max_idx ) = @_;

    my $mid_idx = int( ($max_idx - $first_idx) / 2) + $first_idx;
    my $sort = $cmp->( $sort_key, item_data2sort_key( $array->[$mid_idx] ) ); 
    if ( ($first_idx == $mid_idx && $sort <= 0)  ||
	 ($max_idx  == $mid_idx && $sort >= 0 )  || 
	 $sort == 0
       ) 
    {
	# Search has ended, insert according to sort order
	if ( $sort < 0 ) {
	    # Sort before mid_idx
	    splice( @$array, $mid_idx, 0, $item);
	} else {
	    # Sort right after mid_idx
	    splice( @$array, $mid_idx + 1, 0, $item);
	}
    } else {
	# Recurse
	if ( $sort < 0 ) {
	    binary_insert( $item, $sort_key, $array, $cmp,
			   $first_idx, $mid_idx - 1 );
	} else {
	    binary_insert( $item, $sort_key, $array, $cmp,
			   $mid_idx + 1, $max_idx );
	}
    }
}

sub sort_current_group_element {
    my ( $self, $data ) = @_;

    # Case where @$data is empty
    return unless @$data;

    my $item   = $data->[1];

    # This data item is ended
    my $i = @{$self->group_fields};
    foreach my $op ( @{$self->ops} ) {
	$op->end_group_data( $item->[$i++] );
    }

    if ( $self->{'sort_cmp'}) {
	my $key = item_data2sort_key( $item );
	my $cmp = $self->{'sort_cmp'};
	if ( @$data == 2 ) {
	    push @$data, $item;

	# Small optimization: check for before or at end of array condition
	} elsif ( $cmp->( $key, item_data2sort_key( $data->[2] ) ) <= 0 ) {
	    splice @$data, 2, 0, $item;
	} elsif ( $cmp->( $key, item_data2sort_key( $data->[$#$data])) >= 0 ) {
	    push @$data, $item;
	} else {
	    binary_insert( $item, $key, $data, $cmp, 2, $#$data );
	}
    } else {
	# Push at end
	push @$data, $item;
    }

    # Keep only limit records
    if ( $self->{'limit_num'} ) {
	my $max_count = $self->{'limit_num'} + 2; # last_key, current_element
	splice @$data, $max_count
	  if $max_count < @$data ;
    }
}

# Implements Lire::Aggregator::end_aggregator_data()
sub end_aggregator_data {
    my ( $self, $data ) = @_;

    if ( $self->{'use_array'} ) {
	$self->sort_current_group_element( $data );

	# Remove last_key, current_element
	splice @$data,0,2;
    } else {
	foreach my $key ( keys %$data ) {
	    my $item = $data->{$key};
	    my $i = @{$self->group_fields};
	    foreach my $op ( @{$self->ops} ) {
		$op->end_group_data( $item->[$i++] );
	    }
	}

	# Sort the keys according to the sort value
	my @sorted_keys;
	if ( $self->sort_fields ) {
	    my $cmp = $self->{'sort_cmp'};
	    # This uses schwartzian transform
	    @sorted_keys =  map { $_->[0] } sort $cmp
	      map { [ $_, item_data2sort_key( $data->{$_} ) ] } keys %$data;
	} else {
	    @sorted_keys = keys %$data;
	}

	# Keep only limit records
	if ( $self->{'limit_num'} ) {
	    my $limit = $self->{'limit_num'};
	    splice @sorted_keys, $limit
	      if ($limit < @sorted_keys );
	}

	# Delete unused keys
	%$data = map { $_ => $data->{$_} } @sorted_keys;
	$data->{'_lr_sorted_keys'} = \@sorted_keys;
    }

    $data;
}

# Implements Lire::Aggregator::create_group_entries()
sub create_group_entries {
    my ( $self, $group, $data ) = @_;

    $group->show( $self->{'limit_view'} )
      if $self->{'limit_view'};

    # Either to the sorted group data
    # or the sorted keys
    my $array_ref;
    if ( $self->{'use_array'} ) {
	$array_ref = $data;
    } else {
	$array_ref = $data->{'_lr_sorted_keys'};
    }

    my $field_count = @{$self->group_fields};
    foreach my $elmnt ( @$array_ref ) {
	my $item;
	if ( $self->{'use_array'} ) {
	    $item = $elmnt;
	} else {
	    $item = $data->{$elmnt};
	}

	my $entry = $group->create_entry;
	my $i = 0;
	while  ( $i < $field_count ) {
	    $entry->add_name( $item->[$i++] );
	}
	foreach my $op ( @{$self->ops} ) {
	    $op->add_entry_value( $entry, $item->[$i++] );
	}
    }
}

1;

__END__

=head1 SEE ALSO

 Lire::ReportSpec(3pm), Lire::GroupField(3pm),
 Lire::ReportOperator(3pm)

=head1 AUTHORS

  Francis J. Lacoste <flacoste@logreport.org>
  Wolfgang Sourdeau <wsourdeau@logreport.org>

=head1 VERSION

$Id: Group.pm,v 1.36 2006/07/23 13:16:29 vanbaal Exp $

=head1 COPYRIGHT

Copyright (C) 2001-2004 Stichting LogReport Foundation LogReport@LogReport.org

This file is part of Lire.

Lire 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 2 of the License, or
(at your option) any later version.

This program 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 program (see COPYING); if not, check with
http://www.gnu.org/copyleft/gpl.html.

=cut